本篇我们看一下LangChain中的Agent。在大模型应用开发中，智能体通常指一种以 大语言模型为推理与决策核心 ，结合 记忆 、 工具调用 与环境交互能力，能够进行 规划决策 并执行 复杂任务 以达成目标的软件系统。

Agent的架构如下：

Agent的关键能力

• 理解用户问题
• 如何 拆解任务
• 判断 是否需要工具
• 需要 调用哪些工具
• 如何利用好 工具结果 生成回答&推进任务

1. 新旧版本的agent对比

旧版本的agent

在 LangChain 0.x 时代，框架内的 Agent 系统经历了“碎片化”阶段。当时的设计理念是 “针对场景设计特定 Agent”：

• 如果你要实现思维链推理（ReAct），就用 create_react_agent ；
• 如果需要结构化输出，就用 create_structured_chat_agent ；
• 要工具调用，则用 create_tool_calling_agent 。

python
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 1. 模型初始化
model = ChatOpenAI(model="gpt-4o-mini")
# 2. 创建提示词模板
prompt = PromptTemplate.from_template("""
You are a helpful assistant.
Tools: {tools}
Tool Names: {tool_names}
{agent_scratchpad}
""")
# 3. 创建 agent
agent = create_react_agent(
    llm=model,
    tools=tools,
    prompt=prompt
)
# 4. 创建 executor
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True
)
# 5. 调用
result = executor.invoke({"input": "问题"})

这种方式灵活，但也带来了三个明显问题：

1. 心智负担高——每种 Agent 都要单独记忆 API 与参数；
2. 可组合性差——多个 Agent 之间无法统一调度；
3. 生态碎片化——不同模块难以复用或协同演化。

1.x 版本的agent

LangChain 在 1.0 版本后，团队做出了彻底重构：将所有 Agent 的创建方式统一为一个入口： create_agent()。它取代了旧版本中的 create_react_agent 、 create_json_agent 、create_tool_calling_agent 等多种分支函数，真正让开发者用一行代码即可创建任何类型的智能体。同时在底层通过“中间件机制（Middleware）”和“标准模型接口（invoke/stream）”实现全局统一。这让框架更轻、更稳，也更易于被集成到其他 Agent 平台中。

python
agent = create_agent(
    model=model,
    tools=[tool1, tool2],
    system_prompt="Agent 的行为指令" # 可选
)

result = agent.invoke({"messages": [{"role": "user", "content": "问题"}]})

2. agent 模型的传入方式

在 LangChain 1.2 中， create_agent 是构建智能体的核心方式，底层基于 LangGraph 实现。

create_agent 部分参数如下：

python
from langchain.agents import create_agent

agent = create_agent(
    model: str | BaseChatModel,  # 必需：聊天模型
    tools: List[BaseTool],  # 必需：工具列表
    *,
    system_prompt: str = "",  # 系统提示词
    middleware: Seguence[AgentMiddleware[StateT_co, ContextT]] = ()  # 中间件
    interrupt_before: List[str] = None,  # 在某些工具前暂停（人机协作）
    interrupt_after: List[str] = None,  # 在某些工具后暂停
    debug: bool = False  # 调试模式
    name: str 丨 None = None,  # 设置模型名称
)

Agent在创建时，涉及到 模型(Agent使用的模型) 、 可调用工具 、 系统提示词 等参数的设置。 更多参数参考：https://reference.langchain.com/python/langchain/agents/factory/create_agent

传入字符串

Agent根据传入的模型字符串，自主创建模型对象

python
from dotenv import load_dotenv
from langchain.agents import create_agent

load_dotenv(override=True)
# 只需要传入模型名称，内部自己会从配置文件加载key和base_url
agent = create_agent("deepseek-v4-flash")
print(type(agent))
# <class 'langgraph.graph.state.CompiledStateGraph'>

如果是langchain默认支持的厂商可以这么写，如果langchain没有默认集成的话需要使用下面的方式,比如阿里云百炼下的模型。

传入模型对象

python
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
agent = create_agent(model)
print(type(agent))
from IPython.display import Image, display

display(Image(agent.get_graph().draw_mermaid_png()))

3. 如何调用agent

agent.invoke() 是Agent 最基本的同步调用方法，它会阻塞程序执行直到返回最终结果。具体的：

• 输入：传入的参数为字典类型，字典内通过 messages字段传递消息列表 。即：{"messages": [{"role": "...", "content": "..."}]}
• 输出：通过invoke调用Agent，底层可能会经历多轮交互，返回的是完整的 消息列表 ，被封装在 字典中，是 messages字段的值 。

python
rom rich import print as rprint

# 这里省略了agent初始化的代码
response = agent.invoke({"messages": ["你好"]})  # 默认是HumanMessage
print(type(response))
rprint(response)

4. 如何绑定工具

只有接入了一些工具，create_agent完成Agent创建才算完整。Agent支持 静态 和 动态 绑定工具，后者需要用到中间件，关于中间件我们下篇再讲。 这里的工具可以是LangChain内置的，也可以是自定义的。LangChain生态中已经内置集成了非常多的 实用工具，开发者可以快速调用这些工具完成更加复杂工作流的开发。 LangChain内置工具列表：https://docs.langchain.com/oss/python/integrations/tools

基本用法

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)
agent = create_agent(model)


@tool(parse_docstring=True)
def get_weather(city: str):
    """
    天气查询工具

    Args:
        city: 城市名称
    """
    return f"{city}今天天气挺好"


@tool(parse_docstring=True)
def get_news():
    """
    新闻查询工具
    """
    return "近期，受全球储蓄芯片短缺等多重因素影响，多地回收商称废旧手机回收市场迎来“火热潮”，回收价格普遍上涨，旧手机成“香饽饽”。"


agent = create_agent(
    model,
    tools=[get_weather, get_news]
)
response = agent.invoke({"messages": ["你好，杭州今天的天气如何？今天有哪些新闻？"]})
rprint(response)

5. 重试机制

Agent可以在工具调用结果不满足要求时，自主重试。 示例：

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage
from langchain.tools import tool

load_dotenv(override=True)
flag = 0


@tool
def get_weather(city: str):
    """
    天气查询工具

    Args:
        city: 城市名称
    """
    global flag
    flag += 1
    if flag < 3:
        # raise Exception("暂时无法访问")
        return "TEMP_UNAVAILABLE: 天气服务暂时不可用，请稍后重试"
    return f"{city}今天天气挺好"


model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

messages = [
    SystemMessage("""
        你是一个天气助手。
        当工具返回以 'TEMP_UNAVAILABLE:' 开头的结果时，
        说明是临时故障，不要立即放弃；
        你应再次调用同一个工具，最多重试 3 次。
        如果 3 次后仍失败，再向用户说明服务暂时不可用。
    """),
    HumanMessage("你好，杭州今天的天气如何？")
]

agent = create_agent(model, tools=[get_weather])
response = agent.invoke({"messages": messages})
# rprint(response)
for msg in response["messages"]:
    msg.pretty_print()

6.其它高级用法

设置agent名称

创建Agent时，LangChain允许用户指定其名称

python
agent = create_agent(
    model=model,
    name = "chat_assistant"
)

经典使用场景

name 在 Multi-Agent 场景中最常被提及，用于区分不同的 Agent。但它的作用并不局限于多 Agent编排。在实际工程中，出现如下场景，通常都建议为 Agent 设置一个清晰且稳定的 name 。

1. 流式输出归因

   在启用流式输出时，name 可用于 标识当前输出内容来自哪个 Agent 。 这在多 Agent 协作、Agent 嵌套调用，或前端需要实时展示不同执行主体输出时尤其有用，便 于准确区分 token 或事件的来源。

2. 消息身份标记

   设置 name 后，Agent 产生的 AIMessage 会携带对应的name信息 。 这使得系统在 保存会话记录 、 回放执行过程 、 构建审计日志 或 前端展示消息角色 时，能够明确识别消息的生成者。

3. 调试与trace可读性

   在调试、日志分析和链路追踪过程中，name 可以作为 Agent 的稳定标识，帮助开发者 快速判 断当前执行的是哪个 Agent 。

   当系统中存在多个能力相近的 Agent，或一个 Agent 被嵌套在更复杂的工作流中时，名称能够 显著提升trace 的可读性 和 问题定位 效率。

4. 组件化封装

   在工程实践中，Agent 常被封装为可复用的能力模块，例如检索助手、SQL 助手、报告生成助手 等。为 Agent 设置 name ，有助于在模块注册、运行监控、日志归档和能力复用时保持一致的 身份标识。如果后续需要将该 Agent 进一步作为子图节点、工具能力或子模块接入更复杂系统， 也能降低维护和迁移成本。

5. 作为稳定的运行时身份标识

   从更通用的角度看，name 可以理解为 Agent 在系统中的“ 运行时身份 ID ”。 相比临时性的展示名称，一个稳定、规范的 name 更适合用于 日志检索 、 监控统计 、 链路 分析 和跨模块协作 ，因此在生产环境中通常建议显式设置，而不是依赖默认行为。

系统提示词

使用 create_agent 创建 Agent 时，需传入 模型 和 工具 、可选地传入 系统提示词 。提示词为Agent提供了任务背景、行为准则和操作指南。 系统指令，即SystemMessage，通过 system_prompt 设置，定义 Agent 行为。这个参数可以是 str或者 SystemMessage类型 。

使用建议：

• 明确说明 Agent 的角色
• 定义输出格式
• 说明何时使用工具

python
agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt="""你是天气助手。
工作流程：
1. 理解用户的城市查询
2. 使用 get_weather 工具获取数据
3. 简洁清晰地回答
输出格式：
- 天气状况
- 温度
- 注意事项（如有）
"""
)

提示词设置有两种方式： 静态设置 和 动态设置 。动态设置需要借助中间件，后续讲解。

结构化输出

结构化输出是Agent的核心功能之一，它允许Agent以特定、可预测的格式返回数据，而不是传统的自然语言响应。通过结构化输出，开发者可以直接获得 Pydantic模型 、 JSON对象 或 数据类 等结构化数据，这些数据能够被应用程序直接使用，无需复杂的解析过程。

在上一节我们也研究了下大模型的结构化输出，模型的结构化输出和agent的结构化输出有什么区别呢？

结构化输出的四种策略

LangChain的create_agent()函数自动处理结构化输出的全过程。用户只需通过 response_format参数 设置期望的输出模式（Schema）。 当模型生成结构化数据时，系统会自动捕获、验证并将结果存储在Agent状态的 structured_response键中。

1. ProviderStrategy

   使用模型提供商的 原生结构化输出功能 实现结构化输出。

   这里所说的“原生结构化输出”指的是大语言模型（LLM）提供商通过其API直接提供的、在模型响 应阶段就强制保证 输出格式符合预定规范 的能力，这种能力能够在模型生成内容的源头确保结构 化准确性。

   适用于支持原生结构化输出的模型，比如OpenAI、Anthropic Claude或xAI Grok等。

   python
   from langchain_core.messages import HumanMessage
   from langchain.agents.structured_output import ProviderStrategy
   import os
   from pydantic import BaseModel, Field
   from langchain.agents import create_agent
   from dotenv import load_dotenv
   from langchain.chat_models import init_chat_model
   
   load_dotenv()
   model = init_chat_model(
       model="qwen3.7-plus",
       model_provider="openai",  # 关键：指定使用openai兼容协议
       api_key=os.getenv("DASHSCOPE_API_KEY"),
       base_url=os.getenv("DASHSCOPE_API_BASE_URL")
   )
   
   
   # 2.Pydantic结构化方式定义
   class ContactInfo(BaseModel):
       """用户的联系方式"""
       name: str = Field(description="用户姓名")
       email: str = Field(description="用户邮箱地址")
       phone: str = Field(description="用户的手机号")
   
   
   # 3.agent初始化
   agent = create_agent(
       model=model,
       response_format=ProviderStrategy(ContactInfo)
   )
   
   # 4.调用
   response = agent.invoke({
       "messages": [
           HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：shkstart @ atguigu.com，手机号：12345678912"),
       ]
   })
   # rprint(response)
   for msg in response["messages"]:
       msg.pretty_print()
   

   输出如下：

   ================================ Human Message =================================
   
   从这段话中抽取结构化信息：小明的邮箱地址为：shkstart @ atguigu.com，手机号：12345678912
   ================================== Ai Message ==================================
   
   {"name": "小明", "email": "shkstart@atguigu.com", "phone": "12345678912"}
   

2. ToolStrategy

   对于不支持原生结构化输出的模型，LangChain采用“ToolStrategy”工具调用的方式实现结构化输出。此策略兼容绝大多数 支持工具调用 的现代模型，其核心原理是动态创建一个" 虚拟工具 "，该工具的输入参数对应着期望的数据结构。 当模型需要生成最终答案时，系统会引导模型 "调用"这个虚拟工具 ，从而间接产生符合要求的结构化数据。

   python
   class ContactInfo(BaseModel):
   name: str = Field(description="姓名")
   email: str = Field(description="邮箱")
   phone: str = Field(description="电话")
   
   
   # 3.工具的定义（根据需要定义）
   @tool
   def search_tool(query: str) -> str:
       """
       这是一个搜索引擎。当大模型发现给定的上下文里缺少必要的联系人信息，
       需要去互联网上查询时，才会调用这个工具。
       """
       return f"搜索结果: 未找到关于 '{query}' 的更多额外信息。"
   
   
   # 3.agent初始化
   agent = create_agent(
       model=model,
       tools=[search_tool],
       response_format=ToolStrategy(ContactInfo)
   )
   result = agent.invoke({
       "messages": [{"role": "user", "content": "联系人信息: John Doe,john @ atguigu.com, (010) 56253825"}]
   })
   

   注意：此种方式有些提供商下的模型会报错，例如：qwen3.7-plus就会报错，问了下豆包，它说

   ToolStrategy 配合 create_agent 内部逻辑：自动设置 tool_choice="required"

   DashScope 官方约束：开启思考模式时，不允许 tool_choice=required，仅支持 auto/none

👉 直接触发：InternalError.Algo.InvalidParameter: The tool_choice parameter does not support being set to required or object in thinking mode

可以考虑关闭思考模式

3. type / AutoStrategy

当我们直接传入一个定义类型时，LangChain会自动包装为AutoStrategy，触发 自动选择策略 ：如果模型支持原生结构化输出（如OpenAI、Anthropic Claude或xAI Grok），则优先使用 ProviderStrategy；否则使用ToolStrategy。

python
agent = create_agent(
    model=model,
    tools=[search_tool],
    response_format=AutoStrategy(ContactInfo),
)

这种方式和上面一样，看模型提供商的支持程度，有些会报错

4. None

默认配置，表示不以结构化输出，以 自然语言 响应用户问题。

我们发现不同的模型厂商对response_format参数的支持程度是不一样的，为了解决这个问题，我们可以不使用这个配置参数，agent执行完后，我们再使用model对信息进行二次结构化输出。

python
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from pydantic import BaseModel, Field

load_dotenv()
model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

# 2.Pydantic结构化方式定义
class ContactInfo(BaseModel):
    name: str = Field(description="姓名")
    email: str = Field(description="邮箱")
    phone: str = Field(description="电话")


# 3.工具的定义（根据需要定义）
@tool
def search_tool(query: str) -> str:
    """
    这是一个搜索引擎。当大模型发现给定的上下文里缺少必要的联系人信息，
    需要去互联网上查询时，才会调用这个工具。
    """
    return f"搜索结果: 未找到关于 '{query}' 的更多额外信息。"


# 3.agent初始化
agent = create_agent(
    model=model,
    tools=[search_tool],
    system_prompt="你是联系人提取助手，信息不足调用search_tool查询，收集完信息输出完整联系人资料。"
    # response_format=AutoStrategy(ContactInfo),
)
res_graph = agent.invoke({
    "messages": [{"role": "user", "content": "联系人信息: John Doe,john @ atguigu.com, (010) 56253825"}]
})
print(res_graph)

# 拿到最终文本回答
final_text = res_graph["messages"][-1].content

# 3. 二次结构化解析
structured_model = model.with_structured_output(ContactInfo)
result = structured_model.invoke(final_text)

print(result)
print(f"姓名:{result.name},邮箱:{result.email},电话:{result.phone}")

错误处理：handle_errors参数

受限于模型能力，大模型输出的内容可能并 不符合格式要求 ，ToolStrategy通过其 handle_errors参数 提供了结构化过程错误处理策略，以下是主要的几种方式及其用途：

• handle_errors=True： LangChain默认方式 ， 捕获所有异常 ，并使用LangChain 内置的、信息明确的 错误消息模板 提示模型重试，确保最终能得到符合预定格式的有效数据。适用于大多数 希望自动处理错误的通用场景。

• handle_errors=False：关闭自动重试机制，任何异常都会 直接抛出 ，会 中断程序 运行。

• handle_errors="自定义字符串"：捕获所有异常，但使用开发者 预设的固定字符串 作为错误消 息。适用于需要统一、友好的用户提示，或进行特定业务引导的场景。

• handle_errors=ExceptionType：仅 捕获指定类型(如ValueError) 或元组中的异常类型并进行重试， 其他异常直接抛出 。适用于需要 精准控制 ，只对特定错误进行重试的场景。

• handle_errors=callable：灵活性最高的方式，使用开发者 自定义的函数来处理异常 ，可根据不同的异常类型返回差异化的提示信息。适用于需要复杂、精细化错误处理的场景。

流式输出

通过 invoke 调用Agent时，内部可能经历多次调用，长时间看不到调用情况，用户体验不好，可以通 过 流式调用 （渐进式显示输出）优化用户体验， 实时显示 Agent 运行过程中的更新。特别是在处理LLM 延迟时尤其有效。

通过“ agent.stream(stream_mode=指定模式) ”来指定。具体模式有：values、updates（默认）、messages、custom、checkpoints、tasks、debug。

values输出模式

当 stream_mode 设置为values模式时，每个步骤执行后，都会输出完整的状态信息，适用于每一步都要获取完整状态、状态持久化场景。

完整代码演示：

python
import os
from typing import Dict, Any

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from rich import print as rprint


@tool
def query_customer_data(customer_id: str) -> Dict[str, Any]:
    """
    查询客户基本信息

    Args:
        customer_id: 客户ID，用于唯一标识客户
    Returns:
        包含客户基本信息的字典，如姓名、等级、加入日期等
    """

    # 模拟数据库查询
    return {"name": "张三", "level": "VIP", "join_date": "2023-01-15"}


@tool
def check_order_history(customer_id: str) -> Dict[str, Any]:
    """
    查询客户订单历史

    Args:
        customer_id: 客户ID，用于唯一标识客户
    Returns:
        包含客户订单历史的字典，如总订单数、总花费等
    """

    return {"total_orders": 15, "total_spent": 25800.00}


@tool
def get_current_promotions() -> Dict[str, Any]:
    """
    获取当前可用促销活动

    Returns:
        包含当前可用促销活动的字典，如活动名称、有效日期等
    """
    return {
        "promotions": ["老用户优惠", "会员专属折扣"],
        "valid_until": "2027-01-31"
    }


load_dotenv()

model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

# 创建客户服务Agent
customer_service_agent = create_agent(
    model=model,
    tools=[query_customer_data, check_order_history, get_current_promotions]
)

for chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的个 人信息、历史订单和可用优惠"}]},
        stream_mode="values"
):
    rprint(chunk)
    print("-" * 50)

updates输出模式

这种模式就是默认模式。该模式中，每个步骤执行后，只增量更新状态中发生变化的内容，用于监控 Agent 执行进度，例如观察Agent决定调用工具、工具执行结果等步骤。

python
for chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的个 人信息、历史订单和可用优惠"}]},
        stream_mode="updates"
):
    rprint(chunk)
    print("-" * 50)

messages输出模式

该模式中会输出流式返回的Token以及相关的元数据（如：来自哪个节点），可以用在实现类似 ChatGPT 的打字机效果场景，为聊天机器人等交互式应用提供最佳的实时体验。

python
for chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的个 人信息、历史订单和可用优惠"}]},
        stream_mode="messages"
):
    # rprint(chunk)
    # print("-" * 50)
    print(chunk[0].content, end="", flush=True)

tasks输出模式

该模式会输出当前task任务开始和结束的时间，包含任务的结果和错误信息，该模式用于监控任务的生 命周期。

python
for chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的个 人信息、历史订单和可用优惠"}]},
        stream_mode="tasks"
):
    rprint(chunk)
    print("-" * 50)

debug输出模式

该模式与tasks模式类似，比task模式多输出任务步骤、时间戳、task类型（task/task_result），该模式用于调试、监控task任务的生命周期。

python
for chunk in customer_service_agent.stream(
        {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的个 人信息、历史订单和可用优惠"}]},
        stream_mode="debug"
):
    rprint(chunk)
    print("-" * 50)

checkpoints输出模式

该模式中，每当检查点（checkpoint）被创建时会触发输出，输出包含检查点中的状态，用于需要状态持久化、工作流恢复或分布式执行跟踪的高级场景。

这个模式需要结合上下文的知识，后面等用到了再说，这里只是先演示用法

python
# 1. 创建内存检查点存储
checkpointer = InMemorySaver()
# 2. 创建Agent
customer_service_agent = create_agent(
    model=model,
    tools=[query_customer_data, check_order_history, get_current_promotions],
    checkpointer=checkpointer  # 启用检查点
)
# 3. 创建唯一的会话ID
config = {"configurable": {"thread_id": "session01"}}
# 4. 调用Agent
checkpoint_count = 0
# 使用checkpoints模式进行流式监控
for chunk in customer_service_agent.stream(
        {
            "messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的完整信息和可用优惠"}]
        },
        config=config,
        stream_mode="checkpoints"
):
    checkpoint_count += 1
    print(f"检查点 #{checkpoint_count}")
    print(chunk)
    print("-" * 50)

custom输出模式

开发者通过 get_stream_writer 在工具或节点内部 自定义发送的数据 ，用于 输出 业务逻辑相关的进度信息（如“已处理10/100条记录”）、自定义日志或指标。

举例：生成销售报告和库存报告Agent

python
from langgraph.config import get_stream_writer
from langchain.tools import tool
import time


@tool
def generate_sales_report() -> str:
    """生成销售报告"""

    writer = get_stream_writer()
    writer({"type": "生成销售报告", "message": "开始生成销售报告"})
    # 模拟数据处理
    for i in range(1, 4):
        time.sleep(0.5)
        writer({"type": "生成销售报告", "message": f"生成销售报告进度百分比：{i * 25}%"})

    writer({"type": "生成销售报告", "message": "报告生成完成"})
    return f"销售报告：总收入150万元，同比增长12%"


@tool
def generate_inventory_report() -> str:
    """生成库存报告"""
    writer = get_stream_writer()
    writer("开始库存分析...")
    time.sleep(0.5)
    writer("检查当前库存量...")
    time.sleep(0.5)
    writer("生成库存报告...")
    return "当前库存量为10000件，库存充足，无异常"


# 创建报告生成agent
reporting_agent = create_agent(
    model=model,
    tools=[generate_sales_report, generate_inventory_report]
)
for chunk in reporting_agent.stream(
        {"messages": [{"role": "user", "content": "生成销售报告和库存报告"}]},
        stream_mode="custom"
):
    print(chunk)
    print("-" * 50)

各模式总结

我们可以根据不同的目标来选择不同的输出模式。例如：

• 实现 实时对话交互 ，优先选择messages模式；
• 观察Agent的 思考与执行步骤 ，优先选择updates模式；
• 需要查看 每一步状态 优先选择values/tasks/debug模式；
• 在工具执行时 输出自定义业务 日志优先选择custom模式。

此外，以上这些模式还可以组合使用，例如，可以同时指定 stream_mode=[“tasks”，“updates”] ，这样在同一个循环里 既能 查看Agent task任务执行内容， 又能 显示Agent每步的更新。

python
customer_service_agent = create_agent(
    model=model,
    tools=[query_customer_data, check_order_history, get_current_promotions]
)

for stream_mode, chunk in customer_service_agent.stream(
        {
            "messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的完整信息和可用优惠"}]
        },
        stream_mode=["tasks", "updates"]
):
    print(f"当前流模式: {stream_mode}, 当前数据: {chunk}")
    print("-" * 50)

当指定多模式后，可以通过“ for stream_mode, chunk in customer_service_agent.stream... ”来遍历dict，dict的key(stream_mode)是执行模式，value(chunk)是该模式输出的结果。

实战：多功能助手

项目需求：开发一个多功能智能助手，支持：

1. 天气查询：查询城市天气
2. 数学计算：复杂数学运算
3. 时间查询：获取当前时间、日期计算
4. 货币转换：多种货币之间转换
5. 信息搜索：搜索产品、新闻等信息

第一步： 模型初始化

python
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

model = init_chat_model(
    model="qwen3.7-plus",
    model_provider="openai",  # 关键：指定使用openai兼容协议
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_API_BASE_URL")
)

第二步骤：定义工具

python
from langchain_core.tools import tool


@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的实时天气信息
    支持中国主要城市的天气查询

    Args:
        city: 城市名称，如"北京"、"上海"、"深圳"等
    Returns:
        包含温度、天气状况、空气质量的详细信息
    Examples:
        get_weather("北京") 返回 "多云，15-22℃，空气质量良"
    """

    weather_db = {
        "北京": "多云，15-22℃，空气质量良，湿度 45%",
        "上海": "晴天，18-25℃，空气质量优，湿度 60%",
        "深圳": "小雨，22-28℃，空气质量优，湿度 75%",
        "成都": "阴天，16-23℃，空气质量良，湿度 70%",
        "杭州": "晴天，17-24℃，空气质量优，湿度 55%",
        "广州": "多云，21-29℃，空气质量良，湿度 72%"
    }
    result = weather_db.get(city)
    if result:
        return f"{city}：{result}"
    else:
        return f"抱歉，暂不支持查询{city}的天气信息。当前支持：北京、上海、深圳、成都、杭州、广州"


@tool
def calculator(expression: str) -> str:
    """
    执行数学计算
    支持基本运算符（+、-、*、/、**）和常用数学函数

    Args:
        expression: 数学表达式，可以包含：
            - 基本运算：2 + 3, 10 * 5, 100 / 4
            - 幂运算：2 ** 10
            - 函数：sqrt(16), abs(-5), pow(2, 3)

    Returns:
        计算结果或错误信息

    Examples:
        calculator("2 + 3 * 4") 返回 "14"
        calculator("sqrt(16)") 返回 "4.0"
    """

    try:
        # 安全的数学运算环境
        safe_functions = {
            "sqrt": math.sqrt,
            "pow": pow,
            "abs": abs,
            "round": round,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "log": math.log,
            "pi": math.pi,
            "e": math.e
        }
        result = eval(expression, {"__builtins__": {}}, safe_functions)
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算出错：{str(e)}\n提示：请检查表达式格式，支持的函数有 sqrt,abs, pow, sin, cos, tan, log"


@tool
def get_time_info(query_type: str = "current") -> str:
    """获取时间相关信息

    Args:
        query_type: 查询类型
            - "current": 当前时间
            - "date": 今天日期
            - "tomorrow": 明天日期
            - "yesterday": 昨天日期
            - "weekday": 星期几
    Returns:
        时间信息字符串
    Examples:
        get_time_info("current") 返回 "2025年1月25日 14:30:25"
        get_time_info("weekday") 返回 "星期六"
    """

    now = datetime.now()

    if query_type == "current":
        return now.strftime("当前时间：%Y年%m月%d日 %H:%M:%S")
    elif query_type == "date":
        return now.strftime("今天是：%Y年%m月%d日")
    elif query_type == "tomorrow":
        tomorrow = now + datetime.timedelta(days=1)
        return tomorrow.strftime("明天是：%Y年%m月%d日")
    elif query_type == "yesterday":
        yesterday = now - datetime.timedelta(days=1)
        return yesterday.strftime("昨天是：%Y年%m月%d日")
    elif query_type == "weekday":
        weekdays = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六",
                    "星期日"]
        return f"今天是{weekdays[now.weekday()]}"
    else:
        return f"不支持的查询类型：{query_type}。支持：current, date, tomorrow,yesterday, weekday"


@tool
def convert_currency(amount: float, from_curr: str, to_curr: str) -> str:
    """货币转换工具
    支持主要货币之间的实时汇率转换

    Args:
        amount: 金额数值
        from_curr: 源货币代码（CNY/USD/EUR/GBP/JPY/HKD）
        to_curr: 目标货币代码（CNY/USD/EUR/GBP/JPY/HKD）
    Returns:
        转换结果
    Examples:
    convert_currency(100, "CNY", "USD") 返回 "100 CNY = 14.00 USD"
    """

    # 汇率表（相对于 CNY）
    exchange_rates = {
        "CNY": 1.0,  # 人民币
        "USD": 0.14,  # 美元
        "EUR": 0.13,  # 欧元
        "GBP": 0.11,  # 英镑
        "JPY": 20.8,  # 日元
        "HKD": 1.09  # 港币
    }
    # 货币名称
    currency_names = {
        "CNY": "人民币", "USD": "美元", "EUR": "欧元",
        "GBP": "英镑", "JPY": "日元", "HKD": "港币"
    }

    from_curr = from_curr.upper()
    to_curr = to_curr.upper()

    if from_curr not in exchange_rates:
        return f"不支持的源货币：{from_curr}。支持的货币：CNY, USD, EUR, GBP,JPY, HKD"

    if to_curr not in exchange_rates:
        return f"不支持的目标货币：{to_curr}。支持的货币：CNY, USD, EUR, GBP, JPY, HKD"

    # 转换逻辑：先转为 CNY，再转为目标货币
    cny_amount = amount / exchange_rates[from_curr]
    result_amount = cny_amount * exchange_rates[to_curr]

    from_name = currency_names[from_curr]
    to_name = currency_names[to_curr]

    return f"{amount} {from_name}（{from_curr}）= {result_amount:.2f}{to_name}（{to_curr}）"


@tool
def search_info(keyword: str, category: str = "all") -> str:
    """搜索各类信息

    Args:
        keyword: 搜索关键词
        category: 搜索分类
            - "product": 搜索产品
            - "news": 搜索新闻
            - "all": 搜索所有
    Returns:
        搜索结果
    """

    # 模拟数据库
    products = {
        "手机": "iPhone 15 (¥5999), 小米14 (¥3999), 华为Mate60 (¥6999)",
        "笔记本": "MacBook Pro (¥12999), ThinkPad X1 (¥9999), 华为MateBook(¥7999)",
        "耳机": "AirPods Pro (¥1999), Sony WH-1000XM5 (¥2499)"
    }

    news = {
        "AI": "1. GPT-5 即将发布 2. AI 芯片市场增长 30% 3. 新AI法规出台",
        "科技": "1. 量子计算新突破 2. 6G 技术测试 3. 新能源汽车销量创新高"
    }

    results = []
    if category in ["product", "all"]:
        for key, value in products.items():
            if keyword in key:
                results.append(f"【产品】{key}：{value}")

    if category in ["news", "all"]:
        for key, value in news.items():
            if keyword in key or keyword in value:
                results.append(f"【新闻】{key} 相关：{value}")

    if results:
        return "\n".join(results)
    else:
        return f"未找到关于 '{keyword}' 的{category}信息"

第三步：创建agent

python
from langchain.agents import create_agent


class SmartAssistant:
    """多功能智能助手"""

    def __init__(self):
        # 初始化模型
        self.model = model

        # 工具列表
        self.tools = [
            get_weather,
            calculator,
            get_time_info,
            convert_currency,
            search_info
        ]

        # 系统提示词
        system_prompt = """你是一个多功能智能助手，可以帮助用户：

        🌤 查询天气：使用 get_weather 工具
        🔢 数学计算：使用 calculator 工具
        ⏰ 时间查询：使用 get_time_info 工具
        💱 货币转换：使用 convert_currency 工具
        🔍 信息搜索：使用 search_info 工具

        重要提示：
        1. 仔细阅读用户问题，确定需要使用哪个工具
        2. 如果需要多个工具，按顺序调用
        3. 总是用友好、专业的语气回答
        4. 如果工具返回了数据，要用通俗易懂的语言解释给用户
        5. 如果无法完成任务，诚实地告诉用户原因
        请始终使用中文回答。"""

        # ✅ 创建 agent
        self.agent = create_agent(
            model=self.model,
            tools=self.tools,
            system_prompt=system_prompt
        )

        # 对话历史
        self.messages = []

    def chat(self, user_input: str) -> str:
        """对话接口"""

        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})

        # 调用 agent
        result = self.agent.invoke({"messages": self.messages})

        # 更新消息历史
        self.messages = result["messages"]

        # 返回最后一条 AI 消息
        for msg in reversed(self.messages):
            if msg.type == "ai" and msg.content:
                return msg.content
        return "抱歉，我无法处理这个请求。"

    def reset(self):
        """重置对话历史"""
        self.messages = []

第四步：主程序

python
def main():
    assistant = SmartAssistant()
    print("=" * 40)
    print("🤖 多功能智能助手（LangChain 1.2）")
    print("=" * 40)
    print("\n我可以帮你：")
    print(" 🌤 查询天气")
    print(" 🔢 数学计算")
    print(" ⏰ 时间查询")
    print(" 💱 货币转换")
    print(" 🔍 信息搜索")
    print("\n输入 'quit' 退出，输入 'reset' 重置对话\n")

    demos = [
        "北京今天天气怎么样？",
        "帮我算一下 (25 + 17) * 3",
        "现在几点了？",
        "100 美元等于多少人民币？"
    ]

    for demo in demos:
        print(f"👤 {demo}")
        response = assistant.chat(demo)
        print(f"🤖 {response}\n")

    # 重置对话
    assistant.reset()
    # 交互模式
    print("=" * 40)
    print("💬 进入交互模式")
    print("=" * 40)

    while True:
        user_input = input("\n👤 你: ")
        if user_input.lower() == 'quit':
            print("再见！👋")
            break

        if user_input.lower() == 'reset':
            assistant.reset()
            print("✅ 对话已重置")
            continue

        if not user_input.strip():
            continue

        # 调用助手
        response = assistant.chat(user_input)
        print(f"🤖 助手: {response}")


if __name__ == "__main__":
    main()