Hcc的Blog

嵌入式 · AI · 折腾不止

0%

LangChain 学习(3) Agent 工作流程

发表于 分类于 ，

create_agent() 函数

  create_agent() 是 LangChain 最核心的函数，它会创建一个完整的 Agent 图（StateGraph），包含模型调用、工具执行、循环控制等全部逻辑。

  creat_agent() 函数的语法格式如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain.agents import create_agent

agent = create_agent(
    model,                     # str | BaseChatModel：语言模型
    tools=None,                # Sequence：工具列表
    *,
    system_prompt=None,        # str | SystemMessage：系统提示
    middleware=(),             # Sequence[AgentMiddleware]：中间件列表
    response_format=None,      # ResponseFormat | type：结构化输出配置
    state_schema=None,         # type[AgentState]：自定义状态结构
    context_schema=None,       # type：运行时上下文结构
    checkpointer=None,         # Checkpointer：对话持久化
    store=None,                # BaseStore：跨会话存储
    interrupt_before=None,     # list[str]：在哪些节点前暂停
    interrupt_after=None,      # list[str]：在哪些节点后暂停
    debug=False,               # bool：是否输出详细日志
    name=None,                 # str：Agent 名称
    cache=None,                # BaseCache：缓存配置
)

  其中，tools 参数代表工具列表，它可以接受三种格式的工具：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langchain.tools import tool
from langchain.agents import create_agent

# 格式 1：@tool 装饰的函数（最常用）
@tool
def search_course(keyword: str) -> str:
    """搜索菜鸟教程课程"""
    return f"搜索结果：{keyword} 相关课程"


# 格式 2：Pydantic BaseModel 类
from pydantic import BaseModel, Field

class WeatherQuery(BaseModel):
    """查询天气"""
    city: str = Field(description="城市名称")


# 格式 3：字典（描述远程工具或内置工具）
mcp_tool = {
    "type": "mcp",
    "server_label": "weather_server",
    "server_url": "https://weather.example.com/sse",
    "allowed_tools": ["get_forecast"],
}

# 混合使用
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[search_course, WeatherQuery, mcp_tool],
)

  create_agent() 函数返回一个 CompiledStateGraph 对象，这是 LangGraph 的编译后的图，提供了多种运行方式：

方法 说明 适用场景
invoke(input, config) 同步运行，等待完整结果 脚本、简单接口
aincoke(input, config) 异步运行，等待完整结果 Web 服务
stream(input, config, stream_mode) 同步流式运行 实时展示中间步骤
astream(input, config, stream_mode) 异步流式运行 WebSocket、SSE
get_state(config) 获取当前状态 查看/恢复对话状态
update_state(config, values) 更新状态 手动修改对话状态

LangChain Agent 工作流程

  Agent 的核心是一个简单的循环：调用模型 → 检查是否需要工具 → 执行工具 → 重复。直到模型不再请求工具调用，Agent 停止并返回最终结果。下面我们通过追踪 Agent 的每一步来理解这个过程。新建 py 文件，写入如下代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from model_init import model_init

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。

    Args:
        city: 城市名称
    """
    weather_data = {
        "杭州": "晴，25°C",
        "北京": "多云，18°C",
    }
    return weather_data.get(city, f"未找到 {city} 的天气数据")

@tool
def get_time(city: str) -> str:
    """查询指定城市的当前时间。

    Args:
        city: 城市名称
    """
    time_data = {
        "杭州": "14:30",
        "北京": "14:30",
        "纽约": "02:30",
    }
    return time_data.get(city, f"未找到 {city} 的时间数据")

model = model_init()
agent = create_agent(
    model=model,
    tools=[get_weather, get_time],
    system_prompt="你是一个乐于助人的助手。",
)

# 使用 stream_mode="updates" 可以看到每一个步骤
print("=== Agent 执行过程追踪 ===\n")
step = 0
for chunk in agent.stream(
    {"messages": [HumanMessage(content="杭州现在天气怎么样？几点了？")]},
    stream_mode="updates",
):
    step += 1
    print(f"--- 步骤 {step} ---")
    for node_name, update in chunk.items():
        print(f"节点: {node_name}")
        if "messages" in update:
            for msg in update["messages"]:
                if hasattr(msg, 'tool_calls') and msg.tool_calls:
                    # AI 消息包含工具调用
                    for tc in msg.tool_calls:
                        print(f"  → 请求调用工具: {tc['name']}({tc['args']})")
                elif msg.type == "tool":
                    print(f"  → 工具结果 [{msg.name}]: {msg.content}")
                elif msg.type == "ai" and msg.content:
                    print(f"  → AI 回复: {msg.content[:100]}")

  运行后，程序输出如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
=== Agent 执行过程追踪 ===

--- 步骤 1 ---
节点: model
  → 请求调用工具: get_weather({'city': '杭州'})
  → 请求调用工具: get_time({'city': '杭州'})
--- 步骤 2 ---
节点: tools
  → 工具结果 [get_weather]: 晴，25°C
--- 步骤 3 ---
节点: tools
  → 工具结果 [get_time]: 14:30
--- 步骤 4 ---
节点: model
  → AI 回复: 杭州现在的天气是晴天，气温为25°C，非常舒适。现在的时间是14:30（下午两点半）。

  从这个追踪中可以看到 Agent 执行了 3 个步骤：
  1、Model 节点：模型收到问题，判断出需要调用 get_weather 和 get_time 两个工具，返回两个 tool_call
  2、tools 节点：执行两个工具，获取天气和时间结果
  3、Model 节点：模型收到工具结果，判断信息足够，生成最终回复。

  这里是如何实现 Agent 状态追踪的呢？在 agent 开启流式传输后agent.stream(…) 会返回一个 生成器（generator）。每当你用 for chunk in agent.stream(…) 去遍历时：图里每执行完一个节点（例如模型节点完成一次推理、工具节点完成一次工具调用），生成器就会“吐出”一个新的 chunk。这个 chunk 里包含 该节点造成的状态更新（在 updates 模式下）或 当前完整状态（在 values 模式下）。你的 for 循环会逐块接收这些 chunk，直到整个图运行结束。

  需要注意的是：这里的“流式”和 LLM 生成 Token 级别的流式（stream_mode=”messages”）不太一样：
  updates 模式：以 节点完成 为单位，适合追踪逻辑步骤（调用工具 → 返回结果 → 再推理）。
  messages 模式：以 LLM 生成的 Token 为单位，适合做逐字打字效果。

  stream() 支持多种 stream_mode，每种都提供不同粒度的信息：

模式 返回内容 适用场景
updates 每个节点执行后的状态更新 追踪 Agent 执行步骤，显示中间结果
values 每个节点执行后的完整状态 需要在每一步看到完整消息历史
messages 逐 token 的信息流 前端展示 AI 打字效果
custom 自定义事件 Middleware 通过 stream_writer 发送自定义事件

  那么 Agent 什么时候停止呢？主要有以下几种情况：
  1、无工具调用。当模型返回的 AIMessage 中 tool_calls 为空的时候，模型认为任务完成，直接回复。
  2、return_direct = True。工具标记为直接返回，执行后立即结束，例如查询类工具，结果就是答案。
  3、structured_response，模型产生了结构化输出，response_format 指定的结构化输出完成。
  4、jump_to=”end”，Middleware 通过状态控制主动结束，例如检测到问题越权，提前终止。

LangChain AgentState 状态管理

  Agent 在执行过程中需要维护状态——消息历史、结构化响应、流程控制等。理解 AgentState 的结构和用法，是自定义 Agent 行为的关键。

AgentState 结构

  AgentState 是一个 TypedDict，默认包含三个字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from typing import Annotated
from typing_extensions import Required, NotRequired
from langgraph.graph.message import add_messages
from langgraph.channels.ephemeral_value import EphemeralValue
from langchain.messages import AnyMessage

# AgentState 的实际定义（简化版）
class AgentState(TypedDict):
    # messages：消息历史，使用 add_messages 作为 reducer
    # Required 表示调用时必须提供
    messages: Required[Annotated[list[AnyMessage], add_messages]]

    # jump_to：流程跳转控制，ephemeral（使用后自动清除）
    # NotRequired 表示可选
    jump_to: NotRequired[Annotated[str | None, EphemeralValue]]

    # structured_response：结构化输出结果
    # NotRequired 表示可选，仅在 response_format 设置时出现
    structured_response: NotRequired[Any]

  其中包含如下字段
  messages，类型为 list[AnyMessage]，为必填项，是消息历史，可以通过 add_messages reducer 追加。
  jump_to，类型为 str 或 None，非必填项，是流程跳转控制。
  structured_response，类型为 Any，非必填项，作用为结构化输出结果，不在 input schema 中暴露。

LangChain 提示词

  LangChain 中的提示词有 System Prompt 与 Dynamic Prompt 两种。

system_prompt

  System Prompt 是系统提示词，是控制 Agent 行为的核心手段。create_agent() 的 system_prompt 参数接受两种形式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

# 方式 1：字符串（最简单）
agent = create_agent(
    model=model,
    system_prompt="你是菜鸟教程 RUNOOB 的学习顾问，回答要简洁专业。",
)

# 方式 2：SystemMessage 对象（可复用）
system_msg = SystemMessage(
    content="你是菜鸟教程 RUNOOB 的学习顾问，回答要简洁专业。"
)
agent = create_agent(model=model, system_prompt=system_msg)

  一个好的 system_prompt 应该如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool


@tool
def search_course(keyword: str) -> str:
    """在菜鸟教程搜索课程"""
    courses = {
        "python": "Python3 基础教程（免费）",
        "html": "HTML 基础教程（免费）",
    }
    return courses.get(keyword.lower(), "未找到相关课程")


# 一个设计良好的 system_prompt
GOOD_PROMPT = """你是菜鸟教程 RUNOOB 的学习顾问。

## 你的职责
- 帮助用户找到合适的编程课程
- 回答编程学习相关的问题
- 根据用户水平推荐学习路径

## 行为准则
- 回答要简洁，每次不超过 3 句话
- 优先使用 search_course 工具查询课程信息
- 如果用户是零基础，优先推荐入门课程
- 不使用 emoji 表情
- 不知道的就说不知道，不要编造"""

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[search_course],
    system_prompt=GOOD_PROMPT,
)

result = agent.invoke({
    "messages": [HumanMessage(content="我零基础，想学编程，推荐什么？")]
})
print(result["messages"][-1].content)

@dynamic_prompt ———— 动态生成提示词

  静态 system_prompt 对所有用户一视同仁，但在实际应用中，你可能需要根据用户信息、对话上下文、时间等动态调整提示词。
  @dynamic_prompt 装饰器让你在每次模型调用前动态生成 system_prompt。示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# @dynamic_prompt 装饰器：接收 ModelRequest，返回新的 system_prompt
@dynamic_prompt
def personalized_prompt(request: ModelRequest) -> str:
    """根据对话上下文动态生成个性化提示词"""
    messages = request.state.get("messages", [])
    message_count = len(messages)

    # 可以根据不同的条件动态调整提示词
    base_prompt = "你是菜鸟教程 RUNOOB 的学习顾问。"

    if message_count <= 2:
        # 对话刚开始，耐心引导
        return base_prompt + (
            "用户刚开始对话，请先热情问候，"
            "然后询问他们的学习目标和当前水平。"
        )
    elif message_count > 10:
        # 长对话，提醒保持简洁
        return base_prompt + (
            "对话已经比较长了，回答要尽量简洁，"
            "每次不超过 2 句话。"
        )
    else:
        # 正常对话阶段
        return base_prompt + (
            "根据用户之前的问题推荐合适的课程，"
            "使用 search_course 工具查询课程信息。"
        )

agent = create_agent(
    model=model,
    tools=[search_course],
    middleware=[personalized_prompt],  # 通过 middleware 注入
)

  @dynamic_prompt 的 request 参数提供了丰富的信息，我们可以通过这些信息来获得更好的动态提示词：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from datetime import datetime
from langchain.agents.middleware import dynamic_prompt
from langchain.agents.middleware.types import ModelRequest


@dynamic_prompt
def context_aware_prompt(request: ModelRequest) -> str:
    """根据用户信息、时间和对话阶段动态生成提示词"""
    # 从 runtime.context 获取用户信息
    context = request.runtime.context
    user_name = context.get("user_name", "同学") if context else "同学"
    user_level = context.get("user_level", "入门") if context else "入门"

    # 获取当前时间
    now = datetime.now()
    greeting = "早上好" if now.hour < 12 else "下午好" if now.hour < 18 else "晚上好"

    # 获取当前消息数
    messages = request.state.get("messages", [])

    prompt = f"""你是菜鸟教程 RUNOOB 的学习顾问。

当前时间：{now.strftime('%Y年%m月%d日 %H:%M')}
用户信息：{user_name}，{user_level} 级别

## 行为准则
- 称呼用户为"{user_name}"
- 根据用户级别（{user_level}）推荐合适难度的课程
- 回答要友好但不啰嗦"""

    # 长对话时追加简化提示
    if len(messages) > 20:
        prompt += "\n- 对话很长了，回答尽量精简"

    return prompt

  值得注意的是，@dynamic_prompt 在每次模型调用前都会执行，所以提示词可以随对话推进而变化，但不要在里面做太重的计算，否则会影响响应速度。

  System Prompt 必须包含以下内容：
  角色定义：明确 AI 的身份和职责。
  行为准则：约束回复的风格和边界。
  工具使用指引：告诉模型何时使用哪些工具。
  边界约束：明确什么能做、什么不能做。
  格式要求：指定回复的格式。