从零构建LangChain AI Agent：工具链·记忆环·异常恢复

pip install langchain==0.3.14 langchain-openai==0.3.6 duckduckgo-search==7.5.0 python-dotenv==1.1.0

OPENAI_API_KEY=sk-xxxx
OPENAI_MODEL=gpt-4o
MAX_ITERATIONS=5          # 最大循环次数
MEMORY_TOKEN_LIMIT=2000   # 记忆窗口token上限

import json
import traceback
from langchain_core.tools import tool
from duckduckgo_search import DDGS

class SafeTool:
    """给工具添加统一异常捕获"""
    @staticmethod
    def safe_call(func):
        def wrapper(*args, **kwargs):
            try:
                result = func(*args, **kwargs)
                return result[:2000] if isinstance(result, str) else result
            except Exception as e:
                error_info = {
                    "error": str(e),
                    "traceback": traceback.format_exc()
                }
                return f"【工具错误】{json.dumps(error_info, ensure_ascii=False)}"
        return wrapper

@tool
@SafeTool.safe_call
def web_search(query: str) -> str:
    """搜索互联网信息，输入查询词，返回前3条结果摘要"""
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=3))
    return "\n".join([f"{r['title']}: {r['body'][:200]}" for r in results])

@tool
@SafeTool.safe_call
def python_exec(code: str) -> str:
    """执行Python代码（沙箱环境），返回print输出或错误"""
    import sys
    from io import StringIO
    # 安全限制：禁止文件IO和网络请求
    safe_globals = {"__builtins__": {"print": print, "range": range, "len": len, "str": str, "int": int, "float": float, "list": list, "dict": dict, "tuple": tuple}}
    old_stdout = sys.stdout
    sys.stdout = StringIO()
    try:
        exec(code, safe_globals)
        output = sys.stdout.getvalue()
        return output if output else "代码执行无输出"
    except Exception as e:
        return f"执行异常: {str(e)}"
    finally:
        sys.stdout = old_stdout

tools = [web_search, python_exec]

from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.prompts import ChatPromptTemplate

# 初始化LLM（设置temperature=0.1保持决策稳定）
llm = ChatOpenAI(model="gpt-4o", temperature=0.1, max_tokens=1000)

# 构建提示模板（关键：告诉Agent遇到错误要重试）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以调用工具完成任务。如果工具返回错误信息，尝试换一种方式重试，最多重试2次。"),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 记忆：设置token限制
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    max_token_limit=2000  # 超过时自动丢弃旧消息
)

# 创建Agent
agent = create_openai_functions_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

# 自定义AgentExecutor（增加循环控制和错误处理）
class RobustAgentExecutor(AgentExecutor):
    def __init__(self, max_iterations=5, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_iterations = max_iterations

    def run_with_retry(self, user_input: str) -> str:
        current_input = user_input
        for attempt in range(self.max_iterations):
            try:
                response = self.invoke(
                    {"input": current_input},
                    config={"callbacks": []}
                )
                output = response.get("output", "")
                # 检查输出是否包含Final答案标记
                if "Final Answer:" in output:
                    return output.split("Final Answer:")[-1].strip()
                # 否则把当前输出作为下一轮输入（继续循环）
                current_input = f"基于之前的输出继续：{output}"
            except Exception as e:
                error_msg = str(e)
                # 如果是速率限制，等待1秒重试
                if "rate limit" in error_msg.lower():
                    import time
                    time.sleep(1)
                    continue
                # 其他错误：返回错误信息
                return f"Agent执行失败: {error_msg}"
        return "达到最大迭代次数，任务未完成"

# 实例化
agent_executor = RobustAgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    max_iterations=5,
    verbose=True  # 开启日志
)

if __name__ == "__main__":
    tasks = [
        "搜索'2024年全球GDP排名前三的国家'，然后用Python计算它们GDP总和",
        "我改主意了，把中国换成日本，重新计算"
    ]
    for task in tasks:
        print(f">> 用户: {task}")
        result = agent_executor.run_with_retry(task)
        print(f"> 用户: 搜索'2024年全球GDP排名前三的国家'，然后用Python计算它们GDP总和
[1] 调用 web_search(query='2024年全球GDP排名前三的国家')
[1] 返回: 1. 美国: 28.78万亿美元 2. 中国: 18.53万亿美元 3. 德国: 4.92万亿美元
[2] 调用 python_exec(code='usa=28.78; china=18.53; germany=4.92; total=usa+china+germany; print(f"总和: {total}万亿美元")')
[2] 返回: 总和: 52.23万亿美元
> 用户: 我改主意了，把中国换成日本，重新计算
[3] 调用 web_search(query='2024年日本GDP')
[3] 返回: 日本2024年GDP为4.21万亿美元
[4] 调用 python_exec(code='usa=28.78; japan=4.21; germany=4.92; total=usa+japan+germany; print(f"调整后总和: {total}万亿美元")')
[4] 返回: 调整后总和: 37.91万亿美元
<< Agent: 调整后（美国、日本、德国）的GDP总和为37.91万亿美元。