用 Python 从零写一个 LLM Agent

import json

def run_agent(user_query, llm_call, tools):
    messages = [{"role": "user", "content": user_query}]
    max_turns = 10

    for turn in range(max_turns):
        response = llm_call(messages)

        action = parse_action(response)
        if action["type"] == "answer":
            return action["content"]

        if action["type"] == "tool_call":
            tool = tools.get(action["tool_name"])
            if not tool:
                messages.append({"role": "user", "content": f"Error: tool '{action['tool_name']}' not found."})
                continue
            try:
                result = tool(action["tool_args"])
                messages.append({"role": "user", "content": f"Tool result: {result}"})
            except Exception as e:
                messages.append({"role": "user", "content": f"Tool error: {e}"})

    return "Agent stopped: max turns reached."

def parse_action(text):
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"type": "answer", "content": text}

from openai import OpenAI
client = OpenAI()  # 自动读 OPENAI_API_KEY 环境变量

def call_llm(messages):
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0  # Agent 场景建议用 0，让决策更稳定
    )
    return resp.choices[0].message.content

你是一个可以调用工具的 AI。请始终用 JSON 格式回复。

如果你需要调用工具，返回：
{"type": "tool_call", "tool_name": "calculator", "tool_args": {"expression": "3 + 5 * 2"}}

如果你可以直接回答，返回：
{"type": "answer", "content": "你的回答"}

# 第 1 轮
→ LLM 收到: "北京明天热不热"
← LLM 返回: {"type": "tool_call", "tool_name": "get_weather", "tool_args": {"city": "北京"}}
→ 执行 get_weather("北京")  →  返回: {"temp": 26, "condition": "晴"}

# 第 2 轮
→ 把工具结果喂回 LLM: "Tool result: {"temp": 26, "condition": "晴"}"
← LLM 返回: {"type": "answer", "content": "北京明天晴，气温 26 度，不算热。"}
→ 循环结束，返回给用户

tool_definitions = {
    "get_weather": {
        "description": "查询指定城市的当前天气。只在用户明确询问天气时使用。",
        "parameters": {
            "city": {"type": "string", "description": "城市名称，如 'Beijing'"}
        }
    },
    "search_web": {
        "description": "搜索互联网获取实时信息。只在 LLM 训练数据中不包含该信息时使用。",
        "parameters": {
            "query": {"type": "string", "description": "搜索关键词"}
        }
    },
    "calculator": {
        "description": "执行数学计算。只在用户请求数值计算时使用。",
        "parameters": {
            "expression": {"type": "string", "description": "数学表达式，如 '3 + 5 * 2'"}
        }
    }
}

def build_system_prompt(tool_defs):
    tool_desc = "\n".join(
        f"- {name}: {info['description']} 参数: {info['parameters']}"
        for name, info in tool_defs.items()
    )
    return f"""你是一个可以调用工具的 AI。请始终用 JSON 格式回复。

可用工具：
{tool_desc}

如果需要调用工具，返回：
{{"type": "tool_call", "tool_name": "工具名", "tool_args": {{...}}}}

如果可以直接回答，返回：
{{"type": "answer", "content": "你的回答"}}
"""

# 在每次调用 LLM 时，把 system prompt 放在 messages 的最前面
messages = [{"role": "system", "content": build_system_prompt(tool_definitions)}]
messages.append({"role": "user", "content": user_query})

class AgentMemory:
    def __init__(self):
        self.conversations = []
        self.scratchpad = {}

    def add_message(self, role, content):
        self.conversations.append({"role": role, "content": content})

    def get_context(self, max_messages=50):
        return self.conversations[-max_messages:]

    def note(self, key, value):
        self.scratchpad[key] = value

拿到工具返回的结果后，不要直接返回给用户。请先分析结果，提取关键信息，然后给出用户需要的答案。

如果某个工具已经连续被调用 3 次，请停止并给出你能给的最好回答，不要说"我再试一次"。