从理论到实践的最后一步
前面四篇教程分别讲解了 Agent Loop 的基础理论、工具调用、Reflection 和多 Agent 协作。现在，到了将这些知识融会贯通的时候——我们将构建一个完整的 Agent Loop 应用，它能够自动完成从需求理解到最终交付的全流程。
这个应用将模拟一个「AI 研究助手」：用户提出一个研究主题，Agent 自动进行需求分析、任务分解、信息检索、结果验证和报告生成。整个过程无需人工干预，展示 Agent Loop 在实际场景中的完整能力。
应用架构设计
我们的「AI 研究助手」采用五阶段流水线架构：
- 需求接收：理解用户的研究主题，明确范围和深度要求
- 任务分解：将研究主题拆解为多个可执行的子任务
- 逐步执行：逐个完成子任务，收集信息和分析结果
- 结果验证：交叉验证信息的准确性，补充缺失内容
- 最终交付：整合所有结果，生成结构化的研究报告
每个阶段都可能触发反馈循环——如果验证不通过，会回到执行阶段重新收集信息；如果交付不完整，会回到分解阶段重新规划。
技能链拆解：完整 Agent Loop 应用
我们以「完整 Agent Loop 应用」技能链为例，拆解每个节点在端到端流程中的作用：
节点 1：需求接收（sp-15）——这是整个 Agent Loop 的入口。AI 需要从用户的自然语言输入中提取关键信息：研究主题、期望深度、输出格式、特殊要求等。这个节点的输出是一份结构化的需求文档。
节点 2：任务分解（sp-188）——将需求文档转化为可执行的行动计划。好的任务分解应该：每个子任务独立可执行、子任务之间有清晰的逻辑关系、总工作量合理可控。
节点 3：逐步执行（sp-187）——这是 Agent Loop 的核心执行阶段。AI 按照任务计划，逐个完成子任务。每个子任务可能涉及工具调用、信息检索、代码执行等操作。执行过程会记录日志，便于后续验证。
节点 4：结果验证（sp-185）——对所有执行结果进行质量检查。验证维度包括：信息准确性（是否有事实错误）、覆盖完整性（是否遗漏关键点）、逻辑一致性（各部分是否矛盾）。如果验证不通过，会将具体问题反馈给执行节点进行修正。
节点 5：最终交付（sp-122）——将所有验证通过的结果整合为最终交付物。根据用户的需求格式（报告、PPT、代码仓库等），生成符合要求的输出。这个节点还会添加元数据（生成时间、信息来源、置信度评分等）。
完整代码实现
以下是「AI 研究助手」的完整实现代码：
import json
import time
from datetime import datetime
from openai import OpenAI
client = OpenAI(
api_key="your-deepseek-api-key",
base_url="https://api.deepseek.com"
)
class AIResearchAssistant:
"""完整的 Agent Loop 应用：AI 研究助手"""
def __init__(self, max_iterations=10):
self.max_iterations = max_iterations
self.execution_log = []
self.start_time = None
def log(self, phase, message):
"""记录执行日志"""
entry = {
"timestamp": datetime.now().isoformat(),
"phase": phase,
"message": message
}
self.execution_log.append(entry)
print(f"[{phase}] {message}")
def call_llm(self, system_prompt, user_prompt, response_format=None):
"""统一的 LLM 调用接口"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
kwargs = {
"model": "deepseek-chat",
"messages": messages
}
if response_format:
kwargs["response_format"] = response_format
response = client.chat.completions.create(**kwargs)
return response.choices[0].message.content
def phase1_understand(self, user_input):
"""阶段 1：需求接收与分析"""
self.log("需求分析", "开始分析用户需求")
prompt = f"""分析以下用户的研究需求，提取关键信息。
用户输入：{user_input}
请以 JSON 格式输出：
{{
"topic": "研究主题",
"scope": "研究范围（技术/商业/学术等）",
"depth": "期望深度（概述/详细/深入）",
"format": "期望输出格式",
"keywords": ["关键词1", "关键词2"],
"constraints": ["约束条件1", "约束条件2"],
"target_audience": "目标读者"
}}"""
result = self.call_llm(
"你是一位需求分析专家，善于从模糊描述中提取结构化需求。",
prompt
)
requirements = json.loads(result)
self.log("需求分析", f"识别主题：{requirements['topic']}")
return requirements
def phase2_decompose(self, requirements):
"""阶段 2：任务分解"""
self.log("任务分解", "开始分解研究任务")
prompt = f"""根据以下需求，将研究任务分解为 3-5 个可执行的子任务。
需求：{json.dumps(requirements, ensure_ascii=False)}
请以 JSON 格式输出：
{{
"subtasks": [
{{
"id": 1,
"title": "子任务标题",
"description": "详细描述",
"type": "research|analysis|synthesis|verification",
"estimated_tokens": 500,
"depends_on": []
}}
],
"execution_order": [1, 2, 3]
}}"""
result = self.call_llm(
"你是一位项目管理专家，擅长将复杂任务拆解为可执行的子任务。",
prompt
)
plan = json.loads(result)
self.log("任务分解", f"拆解为 {len(plan['subtasks'])} 个子任务")
return plan
def phase3_execute(self, requirements, plan):
"""阶段 3：逐步执行"""
self.log("执行阶段", "开始执行子任务")
results = {}
for task_id in plan["execution_order"]:
task = next(t for t in plan["subtasks"] if t["id"] == task_id)
self.log("执行阶段", f"执行子任务 {task_id}：{task['title']}")
# 收集依赖任务的上下文
context = ""
for dep_id in task.get("depends_on", []):
if dep_id in results:
context += f"\n前置任务 {dep_id} 的结果：\n{results[dep_id]}"
prompt = f"""完成以下研究子任务。
研究主题：{requirements['topic']}
子任务：{task['title']}
描述：{task['description']}
{"上下文：" + context if context else ""}
要求：
1. 提供准确、有深度的信息
2. 引用具体的数据和案例
3. 指出信息的不确定性（如有）
4. 标注信息来源类型（学术论文/行业报告/新闻/官方数据等）"""
result = self.call_llm(
"你是一位资深研究员，善于收集、分析和整合信息。",
prompt
)
results[task_id] = result
self.log("执行阶段", f"子任务 {task_id} 完成（{len(result)} 字符）")
return results
def phase4_verify(self, requirements, results):
"""阶段 4：结果验证"""
self.log("验证阶段", "开始验证结果质量")
prompt = f"""验证以下研究结果的完整性和准确性。
研究主题：{requirements['topic']}
研究结果：
{json.dumps(results, ensure_ascii=False, indent=2)}
请从以下维度评估：
1. 信息准确性（是否有存疑或不一致的内容）
2. 覆盖完整性（是否遗漏了关键方面）
3. 逻辑一致性（各部分之间是否有矛盾）
以 JSON 格式输出：
{{
"verified": true/false,
"overall_score": 0-100,
"issues": [
{{"task_id": 1, "severity": "high/medium/low", "description": "问题描述"}}
],
"missing_topics": ["遗漏的主题"],
"recommendation": "通过/需补充/需重做"
}}"""
result = self.call_llm(
"你是一位严谨的质量审核专家，不放过任何信息缺陷。",
prompt
)
verification = json.loads(result)
if verification["verified"]:
self.log("验证阶段", f"验证通过，评分：{verification['overall_score']}")
else:
self.log("验证阶段", f"发现问题 {len(verification['issues'])} 个")
return verification
def phase5_deliver(self, requirements, results, verification):
"""阶段 5：最终交付"""
self.log("交付阶段", "生成最终报告")
elapsed = time.time() - self.start_time
prompt = f"""基于以下研究结果，生成最终的研究报告。
研究主题：{requirements['topic']}
目标读者：{requirements['target_audience']}
期望格式：{requirements['format']}
验证结果：{json.dumps(verification, ensure_ascii=False)}
研究数据：
{json.dumps(results, ensure_ascii=False, indent=2)}
报告要求：
1. 结构清晰，包含摘要、正文、结论
2. 专业但易懂，适合目标读者
3. 标注关键数据和引用来源
4. 在末尾附上执行统计（耗时、子任务数、质量评分）
执行统计：
- 总耗时：{elapsed:.1f} 秒
- 子任务数：{len(results)}
- 质量评分：{verification['overall_score']}"""
report = self.call_llm(
"你是一位资深的报告撰写专家，善于将研究数据转化为高质量的报告。",
prompt
)
self.log("交付阶段", f"报告生成完成（{len(report)} 字符）")
return report
def run(self, user_input):
"""运行完整的 Agent Loop"""
self.start_time = time.time()
print("=" * 60)
print("AI 研究助手启动")
print("=" * 60)
for iteration in range(self.max_iterations):
print(f"\n--- 第 {iteration + 1} 轮迭代 ---")
# 阶段 1-2：需求分析与任务分解
if iteration == 0:
requirements = self.phase1_understand(user_input)
plan = self.phase2_decompose(requirements)
# 阶段 3：执行
results = self.phase3_execute(requirements, plan)
# 阶段 4：验证
verification = self.phase4_verify(requirements, results)
if verification["verified"]:
# 阶段 5：交付
report = self.phase5_deliver(requirements, results, verification)
print("\n" + "=" * 60)
print("任务完成！")
elapsed = time.time() - self.start_time
print(f"总耗时：{elapsed:.1f} 秒")
print(f"总迭代：{iteration + 1} 轮")
print(f"子任务数：{len(results)}")
print(f"质量评分：{verification['overall_score']}")
print("=" * 60)
return {
"report": report,
"requirements": requirements,
"verification": verification,
"execution_log": self.execution_log,
"elapsed_seconds": elapsed
}
else:
self.log("迭代", "验证未通过，调整计划重新执行")
# 根据验证反馈调整计划
plan = self.phase2_decompose({
**requirements,
"feedback": verification["issues"],
"missing": verification["missing_topics"]
})
return {"error": "达到最大迭代次数，任务未完成"}
# 使用示例
assistant = AIResearchAssistant(max_iterations=10)
result = assistant.run(
"请研究 2024 年 AI Agent 技术的发展现状和未来趋势，"
"重点关注企业级应用场景，面向技术决策者，"
"输出一份结构化的研究报告。"
)
print(f"\n研究报告：\n{result['report']}")运行效果与关键指标
运行上述代码，你将看到 Agent Loop 的完整执行过程：
- 需求分析：提取研究主题、范围、深度、目标读者等信息
- 任务分解：将研究拆解为背景调研、技术分析、应用场景、竞争格局、趋势预测等子任务
- 逐步执行：每个子任务独立执行，输出详细的分析结果
- 结果验证：检查信息准确性、覆盖完整性和逻辑一致性
- 最终交付：生成结构化研究报告，包含执行统计
关键设计要点：
- 日志系统：每个阶段记录执行日志，便于调试和审计
- 反馈循环：验证不通过时自动回到任务分解阶段，重新规划
- 上下文传递：子任务之间通过依赖关系传递上下文，保证信息连贯
- 质量量化：用评分机制量化输出质量，作为停止条件的判断依据
生产化部署建议
要将这个 Agent Loop 应用部署到生产环境，还需要考虑以下方面：
- 异步执行：使用消息队列（如 Redis、RabbitMQ）实现异步任务处理，避免长时间阻塞
- 状态持久化：将 Agent 的执行状态存入数据库，支持断点续传和进度查询
- 并发控制：对 LLM API 调用实施速率限制和重试机制，避免触发限流
- 成本监控：记录每次调用的 token 消耗，设置预算告警
- 人工审核节点：在关键决策点（如最终交付前）加入人工审核环节
- A/B 测试：对比不同提示词策略的效果，持续优化 Agent 表现
总结与展望
通过这五篇教程，我们从 Agent Loop 的基础概念出发，逐步深入到工具调用、Reflection、多 Agent 协作，最终构建了一个完整的 Agent Loop 应用。Agent Loop 不是遥不可及的概念，而是可以立即上手的开发范式。
Agent Loop 的未来令人兴奋：自主编程、自动化科研、智能运维——这些场景正在从想象变为现实。掌握 Agent Loop，你就掌握了构建下一代 AI 应用的核心能力。现在，打开技能链，亲手编排你的第一个 Agent Loop 吧！
想亲手编排这个技能链？
在技能链中打开 →