跳到主要内容

代理系统详解

什么是LangChain代理​

LangChain代理是一个智能决策系统，它结合了语言模型的推理能力和工具的执行能力，能够自主分析问题、制定计划、执行操作并反复优化。

代理架构​

LangChain代理基于**观察-行动-反思（Observe-Act-Reflect）**循环：

创建代理​

基本创建​

from langchain.agents import create_agent

# 最简形式
agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[tool1, tool2, tool3]
)

完整参数​

from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

agent = create_agent(
    # 模型配置
    model=ChatAnthropic(model="claude-sonnet-4", temperature=0.7),
    
    # 工具列表
    tools=[tool1, tool2],
    
    # 系统提示
    system_prompt="""你是一个专业的助手。
    使用提供的工具完成任务。
    如果不确定，询问用户。""",
    
    # 其他配置
    max_iterations=10,         # 最大迭代次数
    early_stopping_method="force",  # 停止方法
)

代理的工作流程​

详细流程​

输入：用户问题
  ↓
初始化状态
  ├─ messages = [用户消息]
  ├─ 工具列表已加载
  └─ 系统提示已设置
  ↓
第一次迭代
  ├─ 调用模型，传入messages + 工具描述
  ├─ 模型分析：是否需要工具？
  ├─ 是 → 生成工具调用 → 执行工具 → получить结果
  ├─ 否 → 生成最终答案 → 返回结果
  ↓
循环（如果需要更多工具调用）
  ├─ 添加工具结果到messages
  ├─ 调用模型（继续推理）
  ├─ 重复工具调用或返回答案
  ↓
停止条件
  ├─ 模型返回最终答案 → 完成
  ├─ 达到max_iterations → 强制停止
  └─ 出现错误 → 错误处理
  ↓
输出：最终答案

代理的三种操作模式​

1. 同步调用（invoke）​

# 阻塞式执行，等待完整结果
result = agent.invoke({
    "messages": [{"role": "user", "content": "查询用户信息"}]
})

print(result["messages"][-1]["content"])  # 获取最终答案

特点：

  • 完整等待所有工具调用完成
  • 适合简单任务或命令行应用
  • 处理简单，但可能导致长时间等待

2. 流式输出（stream）​

# 逐步流式输出结果
for chunk in agent.stream({
    "messages": [{"role": "user", "content": "..."}]
}):
    print(chunk)  # 实时输出每一步

输出示例：

{'type': 'start', ...}
{'type': 'tool_call', 'tool': 'search', 'args': {...}}
{'type': 'tool_result', 'result': '...'}
{'type': 'step', 'thought': '...'}
{'type': 'end', 'output': '...'}

特点：

  • 实时反馈，改善用户体验
  • 能看到代理的思维过程
  • 适合长时间运行的任务

3. 异步调用（ainvoke）​

import asyncio

async def run_agent():
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "..."}]
    })
    return result

# 运行
result = asyncio.run(run_agent())

特点：

  • 非阻塞执行
  • 适合Web应用和高并发场景
  • 可以进行并行处理

代理的状态管理​

状态结构​

from langchain_core.messages import BaseMessage

state = {
    # 核心字段
    "messages": [...],           # 消息历史
    "messages_info": {...},      # 消息元数据
    
    # 中间状态
    "intermediate_steps": [...], # 工具调用和结果
    "iteration": 0,              # 当前迭代次数
    
    # 用户自定义字段
    "user_id": "user123",
    "session_id": "sess456",
}

状态更新​

# 状态通过消息列表传递
messages = [
    # 轮次1
    {"role": "user", "content": "查询天气"},
    {"role": "assistant", "content": "我来查询..."},
    
    # 工具调用
    {"role": "tool", "name": "weather", "content": "晴天25°C"},
    
    # 轮次2
    {"role": "assistant", "content": "天气是..."},
]

工具调用与执行​

工具调用的生成​

当模型决定使用工具时，它会生成一个结构化的工具调用：

# 模型生成的工具调用
tool_call = {
    "id": "call_123",           # 唯一标识
    "name": "get_weather",      # 工具名
    "args": {                   # 参数
        "city": "旧金山",
        "units": "celsius"
    }
}

自动工具执行​

# LangChain自动处理工具执行
agent = create_agent(
    model=model,
    tools=[get_weather, get_forecast],
    # 自动执行工具，不需要手动干预
)

result = agent.invoke({...})  # 自动处理一切

手动工具执行示例​

# 如果需要自定义执行逻辑
def custom_tool_executor(tool_call):
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    
    # 执行前的检查或修改
    if tool_name == "delete_database":
        if not check_permission():
            return "Permission denied"
    
    # 执行工具
    tool_obj = tools_dict[tool_name]
    result = tool_obj.invoke(tool_args)
    
    return result

停止条件​

代理执行会在以下条件下停止：

1. 正常停止​

# 模型生成了最终答案
# 检查模型输出是否不包含工具调用
if not contains_tool_call(model_output):
    return model_output  # 完成

2. 迭代限制​

agent = create_agent(
    model=model,
    tools=tools,
    max_iterations=10  # 最多10次迭代
)

# 当达到10次迭代后，强制返回最后的状态

3. 显式停止​

# 通过特殊标记停止代理
if "[STOP]" in model_output:
    return "Agent stopped"

代理的错误处理​

工具异常处理​

def safe_tool_call(tool, args):
    try:
        result = tool.invoke(args)
        return {"status": "success", "result": result}
    except Exception as e:
        return {
            "status": "error",
            "error": str(e),
            "suggestion": "Please provide correct parameters"
        }

模型异常处理​

from langchain_core.exceptions import OutputParserException

try:
    result = agent.invoke({...})
except OutputParserException as e:
    # 模型输出无法解析
    print(f"Parse error: {e}")
except Exception as e:
    # 其他异常
    print(f"Error: {e}")
    # 可以重试或返回默认响应

高级特性​

1. 中间件扩展​

from langchain.agents.middleware import wrap_tool_call

def custom_middleware(tool_call):
    """自定义工具调用处理"""
    # 在工具执行前做些处理
    if tool_call["name"] == "delete":
        log_audit_trail(tool_call)
    return tool_call

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[wrap_tool_call(custom_middleware)]
)

2. 自定义循环逻辑​

# 使用LangGraph自定义完整的代理循环
from langgraph.graph import StateGraph

graph = StateGraph(State)

# 定义节点
graph.add_node("model", model_node)
graph.add_node("tools", tools_node)

# 定义边和条件
graph.add_conditional_edges("model", route_to_tools_or_end)

# 编译为runnable
agent = graph.compile()

3. 代理与人工在环​

from langchain.agents.middleware import wrap_tool_call

def require_approval(tool_call):
    """需要人工批准的工具"""
    sensitive_tools = ["delete", "modify_user"]
    
    if tool_call["name"] in sensitive_tools:
        # 发起人工审核流程
        approval = request_human_approval(tool_call)
        if not approval:
            raise PermissionError("Tool call rejected")
    
    return tool_call

性能优化​

1. 减少迭代次数​

# 改进系统提示，减少不必要的工具调用
system_prompt = """
直接完成任务，不要重复验证结果。
如果第一次工具调用成功，不要再调用。
"""

2. 工具执行并行化​

# 当有多个独立工具调用时，并行执行
import asyncio

async def parallel_tool_calls(tool_calls):
    tasks = [execute_tool(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks)
    return results

3. 缓存常用查询​

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_weather(city):
    return get_weather(city)

监控和调试​

1. 启用详细日志​

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("langchain.agents")
logger.setLevel(logging.DEBUG)

# 现在可以看到代理的每一步
result = agent.invoke({...})

2. 使用LangSmith追踪​

from langsmith import traceable

@traceable
def run_agent(query):
    return agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })

3. 检查中间步骤​

result = agent.invoke({...})

# 访问中间步骤
for step in result.get("intermediate_steps", []):
    print(f"Tool: {step[0].tool}")
    print(f"Input: {step[0].tool_input}")
    print(f"Output: {step[1]}")

常见问题​

Q: 如何防止代理调用敏感工具？ A: 使用权限检查中间件，在工具执行前进行验证。

Q: 代理何时应该停止迭代？ A: 当模型生成最终答案（不包含工具调用）或达到最大迭代次数。

Q: 如何改进代理的推理质量？ A: 优化系统提示、添加思维链（Chain-of-Thought）、使用更强大的模型。