LangChain 0.3手写Agent循环：工具注册与记忆回溯机制

python==3.11.9
langchain==0.3.7
langchain-openai==0.2.14
openai==1.54.3
pydantic==2.9.2
sqlite-utils==3.37

from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool

class OrderQueryInput(BaseModel):
    order_id: str = Field(description="订单号，格式如SO20231201")
    user_id: str = Field(description="用户唯一标识")

def query_order(order_id: str, user_id: str) -> str:
    """查询订单状态和物流信息"""
    # 模拟真实API调用，实际场景替换为requests.post
    if not order_id.startswith("SO"):
        raise ValueError("非法订单号前缀")
    return f"订单{order_id}状态：已发货，预计3天后到达"

order_tool = StructuredTool.from_function(
    func=query_order,
    name="query_order",
    description="根据订单号和用户ID查物流",
    args_schema=OrderQueryInput
)

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage

class AgentRunner:
    def __init__(self, tools, llm, max_iterations=5):
        self.tools = {t.name: t for t in tools}
        self.llm = llm.bind_tools(tools)  # 关键：绑定工具后LLM输出会带tool_calls
        self.max_iterations = max_iterations
        self.history = []  # 内存态，实际生产用SQLite持久化

    def run(self, user_input: str):
        self.history.append(HumanMessage(content=user_input))
        for step in range(self.max_iterations):
            # 1. 调用LLM，传入完整历史+系统指令
            ai_msg = self.llm.invoke(self._build_context())
            self.history.append(ai_msg)

            # 2. 判断是否有工具调用意图
            if not ai_msg.tool_calls:
                return ai_msg.content  # 最终答案

            # 3. 执行工具调用
            for tc in ai_msg.tool_calls:
                tool_name = tc["name"]
                tool_args = tc["args"]
                if tool_name not in self.tools:
                    # 错误处理：非法工具名反馈给LLM
                    self.history.append(ToolMessage(content=f"工具{tool_name}不存在，可用: {list(self.tools.keys())}", tool_call_id=tc["id"]))
                    continue
                try:
                    # 4. 真正执行，捕获一切异常
                    result = self.tools[tool_name].invoke(tool_args)
                except Exception as e:
                    # 关键：把错误字符串作为观察返回
                    result = f"执行异常: {type(e).__name__}: {str(e)}"
                self.history.append(ToolMessage(content=result, tool_call_id=tc["id"]))
        # 防止死循环，返回最后一条LLM输出
        return self.history[-1].content or "循环超过最大步数，已终止"

def _build_context(self):
    # 取最近6条消息作为滑窗
    window = self.history[-6:] if len(self.history) > 6 else self.history
    # 如果历史超过6条，把更早的压缩成摘要（用LLM做，但这里用简单截断演示）
    if len(self.history) > 6:
        old_msgs = self.history[:-6]
        summary = f"[早期对话摘要] 共{len(old_msgs)}条，涉及订单查询和优惠券计算。"
        system = SystemMessage(content=f"你是客服助手。{summary} 禁止重复查询已确认的信息。")
    else:
        system = SystemMessage(content="你是客服助手。只调用必要的工具。")
    return [system] + window

# 不用StructuredTool，而是直接定义成普通函数
def query_logistics(order_id: str) -> str:
    # 返回完整JSON，不截断
    import json
    return json.dumps({"tracking": [...]}, ensure_ascii=False)

# 手动构造Tool对象
from langchain_core.tools import Tool
logistics_tool = Tool.from_function(
    func=query_logistics,
    name="query_logistics",
    description="查物流轨迹，返回JSON"
)