从零手写一个AI Agent：LangChain与AutoGPT的工具、记忆与循环控制实现细节

while step  str:
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算错误: {e}"

calc_tool = StructuredTool.from_function(
    func=calculator,
    name="calculator",
    description="计算数学表达式，输入必须是纯字符串表达式",
    args_schema=CalcInput
)

class WebInput(BaseModel):
    url: str = Field(description="要摘要的网页URL")

def web_summary(url: str) -> str:
    resp = requests.get(url, timeout=5)
    soup = BeautifulSoup(resp.text, "html.parser")
    text = soup.get_text()[:800]
    return text.replace("\n", " ")

web_tool = StructuredTool.from_function(
    func=web_summary,
    name="web_summary",
    description="获取网页前800字符的纯文本摘要",
    args_schema=WebInput
)

class FileInput(BaseModel):
    path: str = Field(description="本地文件绝对路径")

def read_file(path: str) -> str:
    with open(path, "r", encoding="utf-8") as f:
        return f.read()[:1000]

file_tool = StructuredTool.from_function(
    func=read_file,
    name="read_file",
    description="读取本地文本文件前1000字符",
    args_schema=FileInput
)

tools = [calc_tool, web_tool, file_tool]

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

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, max_tokens=512)
memory = ConversationBufferWindowMemory(k=5, return_messages=True)

SYSTEM_PROMPT = """你是一个Agent，可以调用工具。可用工具：
- calculator: 计算数学表达式
- web_summary: 获取网页摘要
- read_file: 读取本地文件

如果不需要工具，直接回答。如果需要工具，只输出JSON格式：
{"tool": "工具名", "input": "输入参数"}
"""

def run_agent(user_input: str, max_steps: int = 6):
    memory.chat_memory.add_user_message(user_input)
    step = 0
    last_error = None

    while step < max_steps:
        step += 1
        messages = [SystemMessage(content=SYSTEM_PROMPT)] + memory.load_memory_variables({})["history"]
        try:
            response = llm.invoke(messages)
            content = response.content.strip()
        except Exception as e:
            last_error = e
            time.sleep(1.5 ** step)
            continue

        # 尝试解析工具调用
        if content.startswith("{") and "tool" in content:
            import json
            try:
                call = json.loads(content)
                tool_name = call["tool"]
                tool_input = call["input"]
                tool = next((t for t in tools if t.name == tool_name), None)
                if not tool:
                    raise ValueError(f"未知工具: {tool_name}")

                # 工具调用重试
                for attempt in range(3):
                    try:
                        result = tool.run(tool_input)
                        break
                    except Exception as e:
                        last_error = e
                        if attempt == 2:
                            result = f"工具调用失败: {e}"
                        time.sleep(1.5 ** attempt)

                memory.chat_memory.add_ai_message(f"工具 {tool_name} 返回: {result}")
                continue
            except json.JSONDecodeError:
                pass

        # 普通回答
        memory.chat_memory.add_ai_message(content)
        return content

    return f"达到最大步数({max_steps})，最后错误: {last_error}"

print(run_agent("计算 (25+17)*3 等于多少？"))
# 输出: 工具 calculator 返回: 126
# 最终回答: 126