从零实现LangChain风格AI Agent：工具链、记忆与循环控制

pip install langchain==0.2.15 langchain-openai==0.1.9 openai==1.30.0 python-dotenv==1.0.1

User Input → (LLM + Memory + Tools) → Action → Observation → Repeat until Stop

while step  str:
        # 模拟不稳定API：20%概率失败
        if random.random()  str:
        # 模拟RAG搜索
        doc_db = {
            "项目A预算": "项目A的2024年预算为120万元，其中研发占60%",
            "项目B支出": "项目B截至Q3总支出85万元，超出预算10%",
        }
        return doc_db.get(query, f"未找到与'{query}'相关的文档")

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, SystemMessage

class MemoryManager:
    def __init__(self, max_rounds: int = 5, summary_llm: Optional[ChatOpenAI] = None):
        self.max_rounds = max_rounds
        self.messages = []
        self.summary_llm = summary_llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)

    def add_user_message(self, content: str):
        self.messages.append(HumanMessage(content=content))

    def add_ai_message(self, content: str):
        self.messages.append(AIMessage(content=content))

    def get_context(self) -> str:
        # 如果消息超过max_rounds*2（用户+AI各一轮），进行摘要
        if len(self.messages) > self.max_rounds * 2:
            return self._summarize()
        # 否则直接返回完整历史
        return "\n".join([m.content for m in self.messages])

    def _summarize(self) -> str:
        # 使用LLM压缩历史
        history_text = "\n".join([m.content for m in self.messages[-self.max_rounds*2:]])
        prompt = f"请用两句话总结以下对话历史的核心信息，保持关键数字和结论：\n{history_text}"
        summary = self.summary_llm.invoke([HumanMessage(content=prompt)])
        # 压缩后只保留摘要
        self.messages = [SystemMessage(content=f"历史摘要: {summary.content}")]
        return summary.content

import functools
import time
import random

def retry_on_error(max_retries: int = 3, base_delay: float = 1.0):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt  str:
    retried_run = retry_on_error(max_retries=3)(tool._run)
    return retried_run(**tool_input)

from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage, AIMessage
import json
import re

class SimpleAgent:
    def __init__(self, tools: list, llm: ChatOpenAI, max_steps: int = 10):
        self.tools = {tool.name: tool for tool in tools}
        self.llm = llm
        self.max_steps = max_steps
        self.memory = MemoryManager()
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        tool_descriptions = "\n".join([f"- {t.name}: {t.description}" for t in self.tools.values()])
        return f"""你是一个AI助手，可以调用以下工具来完成任务：
{tool_descriptions}

请按以下格式回复：
- 如果需要调用工具，输出：
  Action: 工具名
  Action Input: 参数JSON

- 如果已经完成任务，输出：
  Final Answer: 最终答案

注意：每次只调用一个工具，等待结果后再决定下一步。"""

    def run(self, user_input: str) -> str:
        self.memory.add_user_message(user_input)
        step = 0
        while step < self.max_steps:
            # 构造当前轮次的prompt
            context = self.memory.get_context()
            messages = [
                SystemMessage(content=self.system_prompt),
                HumanMessage(content=f"历史上下文：\n{context}\n\n当前用户问题：{user_input}\n\n请思考下一步操作：")
            ]
            # 调用LLM
            response = self.llm.invoke(messages)
            content = response.content
            print(f"[Step {step+1}] LLM输出: {content[:100]}...")

            # 解析Action或Final Answer
            action_match = re.search(r"Action:\s*(\w+)\nAction Input:\s*(\{.*\})", content, re.DOTALL)
            final_match = re.search(r"Final Answer:\s*(.*)", content, re.DOTALL)

            if final_match:
                answer = final_match.group(1).strip()
                self.memory.add_ai_message(f"Final Answer: {answer}")
                return answer

            if action_match:
                tool_name = action_match.group(1)
                tool_input_str = action_match.group(2)
                try:
                    tool_input = json.loads(tool_input_str)
                except json.JSONDecodeError:
                    self.memory.add_ai_message("错误：无法解析工具参数，请重新输出Action Input为合法JSON")
                    step += 1
                    continue

                if tool_name not in self.tools:
                    self.memory.add_ai_message(f"错误：工具'{tool_name}'不存在，可用工具: {list(self.tools.keys())}")
                    step += 1
                    continue

                # 安全调用工具（含重试）
                try:
                    observation = safe_invoke_tool(self.tools[tool_name], tool_input)
                    self.memory.add_ai_message(f"Action: {tool_name}\nAction Input: {tool_input_str}\nObservation: {observation}")
                except RuntimeError as e:
                    self.memory.add_ai_message(f"工具调用失败（已重试3次）: {str(e)}")
            else:
                self.memory.add_ai_message("错误：回复格式不正确，请包含Action或Final Answer")
            step += 1

        return "错误：已超过最大推理步数（10步），无法完成请求。"