AI Agent 不是"套壳聊天"：从工具调用到执行闭环的工程化拆解，附可运行代码

# tools.py
import json, httpx

TOOLS = [{
  "type": "function",
  "function": {
    "name": "search_docs",
    "description": "检索内部知识库，返回最相关文档片段",
    "parameters": {
      "type": "object",
      "properties": {
        "query": {"type": "string", "description": "检索关键词"},
        "top_k": {"type": "integer", "minimum": 1, "maximum": 10, "default": 3}
      },
      "required": ["query"]
    }
  }
}]

def search_docs(query: str, top_k: int = 3):
    r = httpx.post("http://kb.internal/search", json={"q": query, "k": top_k}, timeout=10)
    r.raise_for_status()
    return r.json()["hits"]

TOOL_IMPL = {"search_docs": search_docs}

# agent_loop.py
import json, openai

def run_agent(goal: str, max_steps: int = 8):
    messages = [
        {"role": "system", "content": "你是任务型Agent。优先调用工具获取事实，不要编造。完成后输出FINAL。"},
        {"role": "user", "content": goal}
    ]
    trace = []
    for step in range(max_steps):
        resp = openai.chat.completions.create(
            model="gpt-4o", messages=messages, tools=TOOLS, tool_choice="auto")
        msg = resp.choices[0].message

        if not msg.tool_calls:                      # 无工具调用 → 认为完成
            trace.append({"step": step, "action": "final", "content": msg.content})
            return msg.content, trace

        messages.append(msg)                        # 记录模型的工具调用意图
        for call in msg.tool_calls:
            name = call.function.name
            args = json.loads(call.function.arguments)
            try:
                result = TOOL_IMPL[name](**args)    # 执行工具
            except Exception as e:
                result = {"error": str(e)}          # 错误回喂给模型自我修正
            trace.append({"step": step, "tool": name, "args": args, "result": result})
            messages.append({"role": "tool", "tool_call_id": call.id,
                             "content": json.dumps(result, ensure_ascii=False)})
    return "达到最大步数，任务未完成", trace

# memory.py
class AgentMemory:
    def __init__(self, max_short=10):
        self.short, self.max_short = [], max_short

    def add(self, role, content):
        self.short.append({"role": role, "content": content})
        if len(self.short) > self.max_short:          # 滑窗淘汰
            evicted = self.short.pop(0)
            self._persist(evicted)                    # 归档到长期记忆

    def recall(self, query, k=3):
        # 从向量库召回相关历史，拼进上下文
        return vector_db.search(query, k=k)

    def _persist(self, item):
        vector_db.upsert(item)

# planner.py
PLAN_PROMPT = """把目标拆成不超过5个可执行子任务，输出JSON:
{"subtasks":[{"id":1,"desc":"...","done":false}]}"""

def reflect(goal, trace, result):
    resp = openai.chat.completions.create(model="gpt-4o", messages=[
        {"role":"system","content":"你是质检员。判断结果是否满足目标，不满足则指出缺什么。"},
        {"role":"user","content":f"目标:{goal}\n执行轨迹:{trace}\n结果:{result}"}
    ])
    return resp.choices[0].message.content   # 返回"PASS"或修改建议

# guardrail.py
def guard(action, args):
    if action == "search_docs" and len(args.get("query","")) > 200:
        raise ValueError("query过长")
    if action in {"delete_user","transfer_money"}:   # 高危操作需人工确认
        return "NEED_HUMAN_APPROVAL"
    return "OK"

def trace_span(trace, step):
    # 每步落结构化日志：耗时、token、工具、结果摘要
    log.info({"step": step, "tokens": step_tokens, "tool": trace[-1].get("tool")})

# eval.py
CASES = [{"goal":"查A产品的退款政策","expect_tool":"search_docs","expect_kw":"7天"}]

def evaluate(agent_fn):
    passed = 0
    for c in CASES:
        ans, trace = agent_fn(c["goal"])
        ok = c["expect_kw"] in ans and any(t.get("tool")==c["expect_tool"] for t in trace)
        passed += ok
    return passed / len(CASES)   # 工具选择正确率 + 答案命中率