LangChain与AutoGPT双核驱动：手写Agent记忆循环与容错机制

python==3.11.4
langchain==0.1.12
langgraph==0.0.20
zep-python==0.10.4
openai==1.12.0
redis==5.0.3  # 用于记忆缓存

┌─────────────────────────────────────────────┐
│                LangGraph StateGraph          │
│  ┌─────────┐    ┌──────────┐    ┌─────────┐ │
│  │  Agent  │───▶│  Tool    │───▶│ Memory  │ │
│  │  Node   │    │  Executor│    │  Node   │ │
│  └─────────┘    └──────────┘    └─────────┘ │
│       ▲               │               │      │
│       └─────── Error Handler ◀────────┘      │
└─────────────────────────────────────────────┘

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional
import requests

class PriceQueryInput(BaseModel):
    product_name: str = Field(description="产品名称，如'iPhone 15'")

class PriceQueryTool(BaseTool):
    name = "price_query"
    description = "查询指定产品在京东、天猫、拼多多的实时价格，返回JSON格式的低价和高价"
    args_schema: Type[BaseModel] = PriceQueryInput

    def _run(self, product_name: str) -> str:
        # 实际项目中这里应该调用真实API，此处模拟
        cache = redis_client.get(f"price:{product_name}")
        if cache:
            return cache.decode()
        resp = requests.get(
            f"https://api.example.com/v1/price?product={product_name}",
            timeout=5
        )
        result = resp.json()
        # 只保留低价/高价，避免Token膨胀
        simplified = {
            "product": product_name,
            "lowest": min(result["prices"]),
            "highest": max(result["prices"]),
            "source": "jq_tmall_pdd"
        }
        redis_client.setex(f"price:{product_name}", 3600, json.dumps(simplified))
        return json.dumps(simplified)

    async def _arun(self, product_name: str) -> str:
        return self._run(product_name)

from zep_python import ZepClient
from langchain.memory import ZepMemory
from langchain.memory.chat_message_histories import ZepChatMessageHistory

# 初始化Zep服务（docker run -p 8000:8000 -e ZEP_STORE_TYPE=postgres getzep/zep:0.10.4）
client = ZepClient(base_url="http://localhost:8000")

history = ZepChatMessageHistory(
    session_id="agent-session-001",
    client=client,
    memory_type="perpetual",  # 关键：长期记忆
    window_size=6,            # 短期滑动窗口
    search_scope="conversation"  # 实体检索范围
)

memory = ZepMemory(
    chat_memory=history,
    return_messages=True,
    memory_key="chat_history",
    input_key="input",
    output_key="output"
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.runnables import RunnableLambda

class AgentState(TypedDict):
    user_input: str
    chat_history: list
    tool_output: Optional[str]
    final_answer: Optional[str]
    error_count: int

# 定义状态转移节点
def agent_think(state: AgentState):
    # 调用OpenAI GPT-4，附带工具描述
    response = llm.invoke([
        SystemMessage(content=SYSTEM_PROMPT),
        HumanMessage(content=state["user_input"]),
        *state["chat_history"]
    ])
    # 检查是否包含"最终结论"标记
    if "FINAL_ANSWER:" in response.content:
        return {"final_answer": response.content.split("FINAL_ANSWER:")[-1]}
    # 否则解析工具调用
    action = parse_tool_call(response.content)
    return {"tool_call": action}

def tool_execute(state: AgentState):
    tool_name = state["tool_call"]["name"]
    tool_args = state["tool_call"]["args"]
    try:
        tool = registry[tool_name]
        result = tool.run(tool_args)  # 同步执行
        return {"tool_output": result, "error_count": 0}
    except Exception as e:
        # 熔断器：连续3次错误则终止
        new_error = state["error_count"] + 1
        if new_error >= 3:
            return {"final_answer": f"工具调用失败3次，终止任务。最后错误: {str(e)}"}
        return {"tool_output": f"TOOL_ERROR: {str(e)}", "error_count": new_error}

def memory_update(state: AgentState):
    # 将本轮交互写入Zep
    memory.chat_memory.add_message(HumanMessage(content=state["user_input"]))
    if state.get("tool_output"):
        memory.chat_memory.add_message(AIMessage(content=state["tool_output"]))
    return {"chat_history": memory.chat_memory.messages[-6:]}  # 只保留窗口

# 构建状态图
graph = StateGraph(AgentState)
graph.add_node("agent", RunnableLambda(agent_think))
graph.add_node("tool", RunnableLambda(tool_execute))
graph.add_node("memory", RunnableLambda(memory_update))

graph.set_entry_point("agent")
graph.add_edge("agent", "tool")
graph.add_edge("tool", "memory")
graph.add_edge("memory", "agent")

# 条件退出
def should_continue(state):
    if state.get("final_answer"):
        return "end"
    # 最大循环次数
    if state["iteration_count"] >= 8:
        return "end"
    return "continue"

graph.add_conditional_edges("agent", should_continue, {
    "continue": "tool",
    "end": END
})