如何构建真正能交付的 AI 智能体

如何构建真正能上线的 AI 代理 2026 年 6 月 25 日 · 10 分钟阅读 · 查看原文 ↗ AI Claude 营销 自动化 大多数人构建的 AI 代理，在演示里能跑，在生产环境里崩。 这篇指南是写给另一种的。 不是那种你截图发帖的。而是那种凌晨 2 点没人盯着也在跑、处理你早就忘了它存在的边界情况、并且在第 4000 次尝试时仍能返回正确答案的。 我要给你精确展示怎么构建一个。从架构到错误处理到评估。真实代码。真实决策。没有不带应用的理论。 收藏这篇。你还会回来的。 **一个代理到底是什么** 在你构建任何东西之前，先杀掉神话。 一个代理不是魔法。它不是一个人格。它不是有感知的。 一个代理是一个回路。 ``` while goal_not_complete: observation = get_current_state() action = model.decide(observation, tools, memory) result = execute(action) update_state(result) ``` 就这样。一个带着工具的模型，跑在一个回路里，直到完成或失败。 其他一切——框架、抽象、品牌名字——都构建在那个回路之上。 理解了回路，你就理解了你将来要构建或调试的每一个代理。 **第 1 部分——先架构** 大多数教程跳过这个，直接上代码。这就是为什么大多数教程产出的代理会崩。 架构是你在写一行代码之前做的那个决定。它决定后面的一切。 **单代理 vs 多代理** 一个单代理用一个目标处理多个工具。更好构建、更好调试，适合大多数问题。 一个多代理系统有专门的代理把工作交接给彼此。更强大，但要做到可靠则难得多。别从这里开始。 先构建单代理。当你有一个真正需要它的具体问题时，再升级到多代理——而不是因为它听起来更唬人。 **每个代理都需要的 3 样东西** - 一个它能评估的目标 - 一些它能用来取得进展的工具 - 一种知道什么时候算完成的方式 如果这三样里任何一样模糊，你的代理就会空转、幻觉完成、或者一直跑到你把它掐断。 在写代码之前，先用大白话写下这 3 样东西。如果你写不出来，你的代理就还没到可以构建的时候。 **第 2 部分——打地基** 我们来构建一个研究代理。它接收一个问题、搜索信息、读来源、综合一个答案，并引用它找到的东西。 真实问题。真实工具。真实失败模式。 **第 1 步：定义你的工具** 工具是模型可以调用的函数。每个工具都需要一个名字、一段模型会读的描述、输入参数，和一个返回值。 ``` import anthropic import requests from typing import Any client = anthropic.Anthropic() tools = [ { "name": "web_search", "description": "Search the web for current information on a topic. Returns a list of results with titles, URLs, and snippets.", "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query" }, "num_results": { "type": "integer", "description": "Number of results to return (1-10)", "default": 5 } }, "required": ["query"] } }, { "name": "read_url", "description": "Read the full text content of a webpage. Use this to get detailed information from a specific URL.", "input_schema": { "type": "object", "properties": { "url": { "type": "string", "description": "The URL to read" } }, "required": ["url"] } }, { "name": "save_finding", "description": "Save an important finding to memory for use in the final answer.", "input_schema": { "type": "object", "properties": { "finding": { "type": "string", "description": "The finding to save" }, "source": { "type": "string", "description": "The URL or source of this finding" } }, "required": ["finding", "source"] } } ] ``` 注意是什么让这些工具变好： - 描述告诉模型*何时*用它们，而不只是*做什么* - 参数有清晰的类型和描述 - 返回值会是可预测、可解析的 糟糕的工具描述产生糟糕的工具调用。这是大多数代理最先崩掉的地方。 **第 2 步：构建执行层** ``` def execute_tool(tool_name: str, tool_input: dict, memory: list) -> Any: if tool_name == "web_search": # Replace with your actual search API results = mock_search(tool_input["query"], tool_input.get("num_results", 5)) return results elif tool_name == "read_url": try: response = requests.get(tool_input["url"], timeout=10) # In production: parse HTML, extract text, truncate to token budget return {"content": response.text[:3000], "status": response.status_code} except requests.Timeout: return {"error": "Request timed out", "url": tool_input["url"]} except Exception as e: return {"error": str(e), "url": tool_input["url"]} elif tool_name == "save_finding": memory.append({ "finding": tool_input["finding"], "source": tool_input["source"] }) return {"saved": True, "total_findings": len(memory)} else: return {"error": f"Unknown tool: {tool_name}"} ``` 返回错误，而不是抛异常。每个工具都应该返回一些模型能读、能恢复的东西。一个异常会崩掉回路。一个描述性的错误让模型能再试一次。 **第 3 部分——代理回路** ``` def run_agent(question: str, max_iterations: int = 15) -> dict: memory = [] messages = [ { "role": "user", "content": f"""Research this question thoroughly and provide a well-sourced answer. Question: {question} Instructions: - Search for relevant information - Read the most promising sources in full - Save your key findings as you go - After gathering enough information, provide a comprehensive answer - Cite your sources""" } ] system_prompt = """You are a research agent. Your job is to find accurate, sourced answers to questions. Work systematically: 1. Start with broad searches to understand the landscape 2. Narrow to the most relevant sources 3. Read full content when a snippet isn't enough 4. Save findings before you forget them 5. Stop searching when you have enough to answer confidently When you have sufficient information, provide your final answer directly without using more tools.""" iteration = 0 while iteration < max_iterations: iteration += 1 print(f"Iteration {iteration}") response = client.messages.create( model="claude-opus-4-6", max_tokens=4096, system=system_prompt, tools=tools, messages=messages ) # Add assistant response to conversation history messages.append({ "role": "assistant", "content": response.content }) # Check if agent is done if response.stop_reason == "end_turn": final_text = next( (block.text for block in response.content if hasattr(block, "text")), "No response generated" ) return { "answer": final_text, "sources": memory, "iterations": iteration } # Process tool calls if response.stop_reason == "tool_use": tool_results = [] for block in response.content: if block.type == "tool_use": print(f" → Calling {block.name}") result = execute_tool(block.name, block.input, memory) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": str(result) }) messages.append({ "role": "user", "content": tool_results }) # Hit max iterations return { "answer": "Research incomplete: reached maximum iterations", "sources": memory, "iterations": iteration, "error": "max_iterations_reached" } ``` 这个回路做对了、而大多数教程跳过的 3 件事： **对话历史被维护。** 模型能看到之前发生的一切。没有这个，每一轮迭代都是无状态的，代理就无法在之前的步骤上继续构建。 **工具结果作为用户消息返回。** 这是正确的 API 模式。工具结果不是助手消息。 **最大迭代数是一个硬上限，不是一个建议。** 一个没有天花板的代理，如果它搞混了，会永远跑下去。40 美元的 API 账单已经发生过了。设上限。 **第 4 部分——失败处理** 代理就是死在这里的。而且教程几乎从不覆盖它。 **你会撞上的 4 种失败模式** - **幻觉完成。** 模型在真正完成之前就说它做完了。修复：在你的系统提示里加一个完成清单。 - **无限搜索回路。** 模型一直搜，而不是综合。修复：跟踪搜索次数，并在 N 次搜索后提示模型朝答案推进。 - **工具错误螺旋。** 一个工具失败，模型再试、再失败、循环。修复：返回描述性错误，并指示模型在 2 次失败后换路。 - **上下文爆炸。** 长工具结果填满上下文窗口，模型退化。修复：截断工具输出，在返回之前总结长内容。 ``` def execute_tool_with_limits( tool_name: str, tool_input: dict, memory: list, attempt_tracker: dict ) -> Any: # Track attempts per tool+input combination tool_key = f"{tool_name}:{str(tool_input)}" attempt_tracker[tool_key] = attempt_tracker.get(tool_key, 0) + 1 if attempt_tracker[tool_key] > 2: return { "error": f"This exact call has failed {attempt_tracker[tool_key]} times. Try a different approach or different parameters.", "tool": tool_name } result = execute_tool(tool_name, tool_input, memory) # Truncate large outputs result_str = str(result) if len(result_str) > 8000: result_str = result_str[:8000] + "\n\n[Content truncated. Use save_finding to preserve key information before it's lost.]" return result_str return result ``` **第 5 部分——评估** 一个你没法度量的代理，是一个你没法改进的代理。 大多数构建者完全跳过评估。他们上线、生产里出了点问题、他们不知道是什么或为什么，然后花一周救火。 在规模化之前就建好评估。 ``` import json def evaluate_research_agent(test_cases: list) -> dict: results = [] for case in test_cases: print(f"Testing: {case['question'][:60]}...") output = run_agent(case["question"]) # Score the result score = { "question": case["question"], "completed": "error" not in output, "has_sources": len(output.get("sources", [])) > 0, "iterations_used": output.get("iterations", 0), "answer_length": len(output.get("answer", "")), } # Check if expected keywords appear in answer if "expected_keywords" in case: keywords_found = sum( 1 for kw in case["expected_keywords"] if kw.lower() in output.get("answer", "").lower() ) score["keyword_coverage"] = keywords_found / len(case["expected_keywords"]) results.append(score) summary = { "total_cases": len(results), "completion_rate": sum(r["completion_rate"] for r in results) / len(results), "avg_iterations": sum(r["iterations_used"] for r in results) / len(results), "source_rate": sum(1 for r in results if r["has_sources"]) / len(results) } return {"results": results, "summary": summary} # Example test cases test_cases = [ { "question": "What is the current state of AI regulation in the EU?", "expected_keywords": ["AI Act", "regulation", "2024"] }, { "question": "How does transformer attention work?", "expected_keywords": ["attention", "query", "key", "value"] } ] ``` 在每一次有意义的改动上都跑这个。对比分数。如果一次改动改善了延迟、却拉低了关键词覆盖，那是一个值得知道的取舍，而不是在生产里才发现。 **第 6 部分——迈向多代理** 你已经构建了一个能跑的单代理。现在——而且只有现在——才考虑多代理。 管用的模式：一个负责规划和委派的编排者（orchestrator），加上执行任务的专门 worker。 ``` def orchestrator(complex_task: str) -> dict: # Orchestrator breaks the task into subtasks plan_response = client.messages.create( model="claude-opus-4-6", max_tokens=1024, system="You are a task planner. Break complex tasks into clear subtasks that specialized agents can execute. Return JSON.", messages=[{ "role": "user", "content": f"Break this into subtasks: {complex_task}\n\nReturn: {{\"subtasks\": [{{\"id\": 1, \"task\": \"...\", \"type\": \"research|write|analyze\"}}]}}" }] ) plan = json.loads(plan_response.content[0].text) results = {} for subtask in plan["subtasks"]: if subtask["type"] == "research": results[subtask["id"]] = run_agent(subtask["task"]) # Add other specialized agents here # Orchestrator synthesizes final = client.messages.create( model="claude-opus-4-6", max_tokens=4096, messages=[{ "role": "user", "content": f"Original task: {complex_task}\n\nSubtask results: {json.dumps(results)}\n\nSynthesize into a final comprehensive output." }] ) return {"output": final.content[0].text, "subtask_results": results} ``` 编排者模式之所以强大，是因为每个 worker 都可以被独立地专门化、评估和改进。当研究代理变好时，整个系统就变好。 **你真正需要什么才能上线** 一组合已上线的代理，胜过一年的教程。 按顺序构建这 3 个： **项目 1。** 一个带评估的单代理研究工具。就是这篇指南里的那个。把它部署到某个真人能用它的地方。度量完成率和来源质量。 **项目 2。** 一个带 RAG 的文档问答代理。接收一文件夹 PDF，回答关于它们的问题，引用页码。加评估：它找得到正确的 chunk 吗？答案跟来源匹配吗？ **项目 3。** 一个解决真实工作流的多代理系统。挑一件你手动做的事，把它自动化。至少一个编排者和 2 个 worker。记录什么坏了、你怎么修的。 3 个部署好的、带文档化评估的项目，是一个能打开大多数 CS 毕业生都打不开的门的作品集。 **没人告诉你的部分** 构建代理最难的部分不是代码。 是知道你的代理什么时候错了。 一个大声失败的代理很容易修。一个自信地返回一个貌似合理、却错误的答案的代理，才是那个让你损失用户、声誉或金钱的。 在规模化之前建好你的评估。度量重要的东西。相信数字，而不是演示。 上线可靠代理的团队，不是那些找到了更好框架的团队。 是那些认定「在演示里能跑」永远不够好的团队。 从那里开始，其他一切都会跟上。 标签：# X # AI # Claude # Marketing # Automation # Growth # Guide 相关文章 How to design an AI agent How I Use Claude Fable 5 to Build $10k/Mo Faceless AI Story Channels The Production AI Stack for Solo Builders