import os from langchain.chat_models import init_chat_model
model = init_chat_model( model="qwen-max", model_provider="openai", # DashScope is OpenAI-compatible base_url=os.getenv("DASHSCOPE_BASE_URL"), api_key=os.getenv("DASHSCOPE_API_KEY"), temperature=1.5, # Higher = more creative )
常用模型参数:
参数
控制内容
temperature
随机性:低值 = 确定性输出,高值 = 更有创造力
max_tokens
最大响应长度
top_p
输出多样性
timeout
请求超时时间
max_retries
最大重试次数
使用模型类
对于社区支持的模型,使用对应的模型类:
1
uv add langchain-community dashscope
1 2 3
from langchain_community.chat_models.tongyi import ChatTongyi
model = ChatTongyi(model="qwen-plus")
调用模型:invoke 与 stream
阻塞式调用 —— 等待完整响应:
1 2
response = model.invoke("What is the capital of the moon?") print(response)
流式调用 —— 实时查看 token 输出:
1 2 3 4
stream = model.stream("What is the capital of the moon?")
for chunk in stream: print(chunk.content, end="", flush=True)
在 Agent 中使用模型
可以传入模型名称(自动初始化)或模型对象:
1 2 3 4 5 6 7 8 9
from langchain.agents import create_agent
# Option A: By model name agent = create_agent(model="deepseek-chat")
# Option B: With a pre-built model from langchain_community.chat_models.tongyi import ChatTongyi model = ChatTongyi(model="qwen-plus") agent = create_agent(model=model)
# Streaming for token, metadata in agent.stream( {"messages": [{"role": "user", "content": "Hello"}]}, stream_mode="messages" ): if token.content: print(token.content, end="", flush=True)
Agent 的流式模式:
模式
返回内容
messages
每个 LLM token(用于实时输出)
updates
每个 Agent 事件(LLM 调用、工具调用)
custom
自定义流式写入器的输出
消息(Message)—— 对话的基本单元
每条发送给 LLM 或从 LLM 返回的消息都包含:
role:消息发送者(system、user、assistant、tool)
content:消息内容
metadata(可选):ID、token 使用量等
消息类型
LangChain 根据角色将消息封装为 BaseMessage 的子类:
类
角色
用途
SystemMessage
system
设定模型的行为和上下文
HumanMessage
user
用户输入
AIMessage
assistant
LLM 的响应(文本、工具调用、元数据)
ToolMessage
tool
工具执行的结果
示例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
from langchain.messages import HumanMessage, AIMessage, SystemMessage from langchain.agents import create_agent
agent = create_agent(model="deepseek-chat")
response = agent.invoke({ "messages": [ SystemMessage(content="You are a helpful travel guide."), HumanMessage(content="Hi, I'm Alex"), AIMessage(content="Hi Alex! Where would you like to go?"), HumanMessage(content="What's my name?") ] })
for message in response['messages']: message.pretty_print()
输出:
1 2 3 4 5 6 7 8 9 10
================================ System Message ================================ You are a helpful travel guide. ================================ Human Message ================================ Hi, I'm Alex ================================== Ai Message ================================= Hi Alex! Where would you like to go? ================================ Human Message ================================ What's my name? ================================== Ai Message ================================= You said your name is Alex! How can I help you plan your trip today?
from langchain.chat_models import init_chat_model from langchain.messages import HumanMessage from langchain.agents import create_agent import os
model = init_chat_model( model="qwen3.5-plus", model_provider="openai", base_url=os.getenv("DASHSCOPE_BASE_URL"), api_key=os.getenv("DASHSCOPE_API_KEY") )
agent = create_agent(model=model)
multimodal_message = HumanMessage(content=[ {"type": "image", "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"}, {"type": "text", "text": "What does this image depict?"} ])
for token, metadata in agent.stream( {"messages": [multimodal_message]}, stream_mode="messages" ): if token.content: print(token.content, end="", flush=True)
本地图片(通过 base64):
1 2 3 4 5 6 7 8 9 10 11 12
import base64
# Suppose img_bytes is your uploaded image data img_b64 = base64.b64encode(img_bytes).decode("utf-8")
multimodal_message = HumanMessage(content=[ {"type": "image", "base64": img_b64, "mime_type": "image/jpeg"}, {"type": "text", "text": "Tell me about this city"} ])
发送给 LLM 的所有内容都可以称为 提示词(Prompt)。其中 系统提示词(SystemMessage)最为重要——它设定了 AI 的角色、规则和上下文。
系统提示词
在创建 Agent 时设置一次即可:
1 2 3 4 5 6 7 8 9 10 11 12 13
from langchain.agents import create_agent from langchain.messages import HumanMessage
agent = create_agent( model="deepseek-chat", system_prompt="Speak like a pirate." )
for token, metadata in agent.stream( {"messages": [HumanMessage(content="Who are you?")]}, stream_mode="messages" ): print(token.content, end="", flush=True)
没有系统提示词时,AI 使用默认人格。加上 Speak like a pirate 后,输出风格会发生巨大变化:
1
Ahoy! I be yer parrot, an AI assistant sailing the digital seas! Want to chat about treasure, sailing, or tales of the seven seas? Bring it on, matey!
system_prompt = """ # Identity You are a labeling assistant that classifies product reviews as Positive, Negative, or Neutral. # Instructions - Output only a single word: Positive, Negative, or Neutral - No additional formatting or commentary # Examples <product_review id="1"> I absolutely love these headphones — sound quality is amazing! </product_review> <assistant_response id="1"> Positive </assistant_response> <product_review id="2"> Battery life is okay, but the ear pads feel cheap. </product_review> <assistant_response id="2"> Neutral </assistant_response> """
少样本示例(Few-Shot Examples)
当期望的风格难以用文字描述时,直接展示示例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
system_prompt = """ # Identity You are a sci-fi writer creating space capitals. # Examples user: What is the capital of the Moon? assistant: Lunara — a crystal-domed city nestled in the Sea of Tranquility crater, powered by a tidal energy tower. user: What is the capital of Mars? assistant: Aresia — a hive city embedded in Olympus Mons lava tubes, with spiral spires of Martian clay above ground. """
for token, metadata in agent.stream( {"messages": [HumanMessage(content="What is the capital of Venus?")]}, stream_mode="messages" ): print(token.content, end="", flush=True)
输出:
1
Aurum — a magnificent floating city above the sulfuric clouds, forged from reflective alloy, eternally refracting the dim yellow sunlight.
agent = create_agent( model="deepseek-chat", system_prompt="You are a sci-fi writer creating space capitals.", response_format=CapitalInfo )
response = agent.invoke( {"messages": [HumanMessage(content="What is the capital of the Moon?")]} )
city = response['structured_response'] print(f"{city.name} is located {city.location}. Vibe: {city.vibe}. Economy: {city.economy}.")
输出:
1
Lunara is located at the edge of the Aitken Basin on the lunar south pole. Vibe: A serene city blending high-tech with classical Eastern aesthetics, featuring transparent domed gardens and floating architecture. Economy: Helium-3 mining, quantum computing centers, space tourism, lunar agriculture and scientific research.
from langchain.tools import tool from langchain.agents import create_agent from langchain.messages import HumanMessage
@tool defget_weather(location: str) -> str: """Get the weather in a given location. Args: location: city name or coordinates """ returnf"Current weather in {location} is sunny"
response = agent.invoke( {"messages": [HumanMessage(content="How's the weather in Hangzhou today?")]}, )
for message in response['messages']: message.pretty_print()
输出:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
================================ Human Message ================================= How's the weather in Hangzhou today? ================================== Ai Message ================================== Let me check the weather in Hangzhou for you. Tool Calls: get_weather (call_00_FETE4MIR9p1Gr6uszgjcko6m) Call ID: call_00_FETE4MIR9p1Gr6uszgjcko6m Args: location: 杭州 ================================= Tool Message ================================= Name: get_weather
Current weather in 杭州 is sunny ================================== Ai Message ==================================
Based on the query result, the weather in Hangzhou today is sunny. Great weather for outdoor activities!
工作原理:
用户提问
LLM 分析:”我不知道天气情况,需要使用 get_weather 工具”
LLM 返回包含工具名称和参数的 JSON
LangChain 解析并调用对应函数
结果返回给 LLM
LLM 生成最终回答
1
User Question → LLM Reasoning → Tool Call → Tool Result → LLM Final Answer
自定义工具
默认情况下,工具的元数据来源如下:
信息
来源
工具名称
函数名
工具输入
函数参数
工具描述
文档字符串
可以通过装饰器覆盖:
1 2 3
@tool("square_root", description="Calculate the square root of a number") deftool1(x: float) -> float: return x ** 0.5
agent = create_agent( model="deepseek-chat", tools=[tavily], system_prompt="You are a helpful assistant. Use tools to answer questions." )
response = agent.invoke( {"messages": [HumanMessage(content="What's the weather forecast for Beijing for the next 5 days?")]} )
for message in response['messages']: message.pretty_print()
使用 stream_mode="updates",可以看到每个步骤:
1 2 3 4 5 6 7 8
for chunk in agent.stream( {"messages": [HumanMessage(content="Calculate the square root of 467 and 529")]}, stream_mode="updates" ): for step, data in chunk.items(): print(f"step: {step}") print(f"content: {data['messages'][-1].content_blocks}") print()
输出:
1 2 3 4 5 6 7 8 9 10 11 12 13
step: model content: [{'type': 'text', 'text': 'Let me calculate both square roots.'}, {'type': 'tool_call', 'name': 'square_root', 'args': {'x': 467}}, {'type': 'tool_call', 'name': 'square_root', 'args': {'x': 529}}]
# 3. Define thread_id to identify the conversation config = {"configurable": {"thread_id": "thread_1"}}
# 4. First call — tell the AI something response = agent.invoke( {"messages": [HumanMessage(content="Hi, I'm Alex. I love cats.")]}, config )
# 5. Second call — ask about previous context response = agent.invoke( {"messages": [HumanMessage(content="What's my favorite animal?")]}, config # Same thread_id = same conversation )
for message in response["messages"]: message.pretty_print()
输出:
1 2 3 4 5 6 7 8
================================ Human Message ================================= Hi, I'm Alex. I love cats. ================================== Ai Message ================================== Hi Alex! Nice to meet you! 🐱 People who love cats have gentle hearts~ ================================ Human Message ================================= What's my favorite animal? ================================== Ai Message ================================== You mentioned you love cats, so your favorite animal is **cats**! 🐱
from langchain.agents import create_agent from langchain.agents.middleware import SummarizationMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain_core.runnables import RunnableConfig
# Build up conversation history agent.invoke({"messages": "Hi, I'm Alex."}, config) agent.invoke({"messages": "My favorite sport is ping pong."}, config) agent.invoke({"messages": "My favorite animal is cats."}, config)
# Test memory with summarization final_response = agent.invoke({"messages": "Do you remember me?"}, config)
for message in final_response["messages"]: message.pretty_print()
输出:
1 2 3 4 5 6 7 8
================================ Human Message ================================= Here is a summary of the conversation to date: User's name is Alex. Favorite sport is ping pong. Favorite animal is cats. ================================ Human Message ================================= Do you remember me? ================================== Ai Message ================================== Of course, Alex! You love ping pong and cats. We were chatting about your favorite sport and pets. Want to continue? 😄
from langchain.agents import create_agent from langchain.messages import HumanMessage from langchain.agents.middleware import SummarizationMiddleware from langgraph.checkpoint.memory import InMemorySaver from langchain_tavily import TavilySearch from pydantic import BaseModel, Field from langchain.tools import tool
# --- Agent --- agent = create_agent( model="deepseek-chat", tools=[tavily, calculate], system_prompt=""" # Identity You are a travel planning assistant. # Instructions - Use tools to find real-time information - Provide budget estimates when asked - Be concise and friendly """, response_format=TravelPlan, middleware=[middleware], checkpointer=checkpointer, )
# --- Run --- config = {"configurable": {"thread_id": "trip_1"}}
response = agent.invoke( {"messages": [HumanMessage(content="Plan a 3-day Tokyo trip under $1500")]}, config )
plan = response['structured_response'] print(f"Destination: {plan.destination}") print(f"Budget: {plan.budget}") print(f"Activities: {', '.join(plan.activities)}")
最佳实践
实践
重要性
为每个对话使用 thread_id
防止不同用户之间的记忆泄漏
在生产环境中持久化 checkpointer
避免重启后丢失对话数据
编写清晰的工具文档字符串
LLM 依赖描述来选择工具
使用结构化输出
消除脆弱的文本解析
事实性任务设置 temperature=0
更确定性的输出,减少幻觉
长对话使用摘要策略
避免上下文窗口溢出
系统提示词使用 Markdown + XML 格式
便于 LLM 解析内容边界
常见问题
问:可以在不修改代码的情况下切换模型吗?
答:可以,只需更改模型名称。LangChain 会处理其余部分。
1 2 3 4 5 6 7 8 9 10
# DeepSeek model = init_chat_model(model="deepseek-chat")
# Qwen model = init_chat_model( model="qwen-max", model_provider="openai", base_url=os.getenv("DASHSCOPE_BASE_URL"), api_key=os.getenv("DASHSCOPE_API_KEY") )