自主智能体端到端闭环实战工具全拆解

graph TD
    A[用户目标输入] --> B[意图解析与任务分解]
    B --> C[执行规划器 Planner]
    C --> D[工具选择器 Tool Router]
    D --> E{工具执行}
    E --> F[结果观察器 Observer]
    F --> G{目标达成?}
    G -->|否| H[反思与修正 Reflector]
    H --> C
    G -->|是| I[结果合成与交付]

    subgraph 持久化层
        J[工作记忆 Working Memory]
        K[长期记忆 Long-term Memory]
        L[状态检查点 Checkpoint]
    end

    C <--> J
    H <--> K
    E <--> L

    subgraph 安全护栏
        M[权限边界]
        N[成本控制]
        O[输出审计]
    end

    D --> M
    E --> N
    I --> O

from agents import Agent, Runner, function_tool

@function_tool
def search_codebase(query: str, file_pattern: str = "*.py") -> str:
    """搜索代码库中的相关文件和函数"""
    # 实际实现连接到代码索引服务
    import subprocess
    result = subprocess.run(
        ["grep", "-rn", query, "--include", file_pattern, "."],
        capture_output=True, text=True, timeout=30
    )
    return result.stdout[:3000] if result.stdout else "未找到匹配结果"

@function_tool
def run_tests(test_path: str = "tests/") -> str:
    """运行测试套件并返回结果"""
    import subprocess
    result = subprocess.run(
        ["pytest", test_path, "-v", "--tb=short"],
        capture_output=True, text=True, timeout=120
    )
    return result.stdout[-3000:] if result.stdout else result.stderr[-3000:]

code_agent = Agent(
    name="代码助手",
    instructions="""你是一个专业的代码分析和修改助手。
    当用户提出代码相关请求时：
    1. 先搜索相关代码
    2. 理解现有实现
    3. 提出修改方案
    4. 执行修改后运行测试验证""",
    tools=[search_codebase, run_tests],
    model="gpt-4.1"
)

result = await Runner.run(
    code_agent,
    input="重构utils.py中的所有函数，添加类型注解",
    max_turns=25
)
print(result.final_output)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    iteration_count: int
    token_budget: int
    tools_used: list[str]
    reflection_notes: list[str]

def planner(state: AgentState) -> dict:
    """规划下一步行动"""
    if state["iteration_count"] > 10:
        return {"messages": [{"role": "assistant", 
                 "content": "已达到最大迭代次数，总结现有结果。"}]}
    if state["token_budget"] < 2000:
        return {"messages": [{"role": "assistant",
                 "content": "Token预算不足，切换到精简模式。"}]}
    # 调用LLM进行规划
    response = llm.invoke(state["messages"])
    return {"messages": [response], "iteration_count": state["iteration_count"] + 1}

def should_continue(state: AgentState) -> str:
    """条件路由：决定是否继续循环"""
    last_msg = state["messages"][-1]
    if last_msg.get("tool_calls"):
        return "execute_tools"
    if state["iteration_count"] > 10:
        return "summarize"
    return "reflect"

def reflect(state: AgentState) -> dict:
    """反思节点：评估当前进展"""
    reflection = llm.invoke([
        {"role": "system", "content": "分析当前进展，决定是否需要调整策略。"},
        {"role": "user", "content": f"已执行{len(state['tools_used'])}次工具调用。"
         + f"反思笔记：{state['reflection_notes']}"}
    ])
    return {
        "reflection_notes": [reflection.content],
        "messages": [reflection]
    }

graph = StateGraph(AgentState)
graph.add_node("planner", planner)
graph.add_node("execute_tools", tool_executor)
graph.add_node("reflect", reflect)
graph.add_node("summarize", summarizer)

graph.add_conditional_edges("planner", should_continue, {
    "execute_tools": "execute_tools",
    "reflect": "reflect",
    "summarize": "summarize"
})
graph.add_edge("execute_tools", "planner")
graph.add_edge("reflect", "planner")
graph.add_edge("summarize", END)

graph.set_entry_point("planner")
app = graph.compile()

# 检查点持久化——中断后可恢复
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

result = app.invoke(
    {"messages": [{"role": "user", "content": "调研量子计算在密码学中的应用"}],
     "iteration_count": 0, "token_budget": 50000,
     "tools_used": [], "reflection_notes": []},
    config={"configurable": {"thread_id": "research-001"}}
)

from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool, ScrapeElementFromWebsiteTool

llm = LLM(model="gpt-4.1", temperature=0.1)

researcher = Agent(
    role="高级市场研究员",
    goal="深度调研指定领域的市场动态、竞争格局和技术趋势",
    backstory="""你是一位在咨询公司工作10年的市场研究专家。
    你擅长从海量信息中提取关键洞察，并结构化呈现。
    你从不猜测——所有结论都有数据支撑。""",
    tools=[SerperDevTool(), ScrapeElementFromWebsiteTool()],
    llm=llm,
    verbose=True,
    max_iter=5,
    memory=True
)

analyst = Agent(
    role="数据分析师",
    goal="将原始数据转化为可执行的商业洞察",
    backstory="你是前麦肯锡分析师，擅长数据建模和趋势预测。",
    tools=[],
    llm=llm,
    verbose=True,
    memory=True
)

writer = Agent(
    role="技术写作专家",
    goal="将分析结果转化为清晰有力的报告",
    backstory="你是前《MIT Technology Review》编辑。",
    tools=[],
    llm=llm,
    verbose=True
)

research_task = Task(
    description="""调研2026年AI Agent市场的以下方面：
    1. 市场规模与增长率
    2. 前5大玩家及市场份额
    3. 关键技术趋势
    4. 用户采用障碍
    输出结构化的调研报告。""",
    expected_output="包含数据支撑的市场调研报告，至少2000字",
    agent=researcher
)

analysis_task = Task(
    description="""基于调研数据，进行以下分析：
    1. 市场机会评估
    2. 竞争优势分析框架
    3. 进入市场策略建议""",
    expected_output="战略分析报告，包含可执行建议",
    agent=analyst
)

report_task = Task(
    description="""将调研和分析结果整合为一份面向CTO的决策简报。
    要求：简洁有力、数据驱动、有明确的行动建议。""",
    expected_output="CTO决策简报，1500字以内",
    agent=writer
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.hierarchical,  # Manager自动编排
    verbose=True,
    memory=True
)

result = crew.kickoff()
print(result.raw)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1")

# 自主代码开发Agent——可以自己写代码、运行、调试
coder = AssistantAgent(
    name="coder",
    model_client=model_client,
    tools=[execute_code_tool, search_documentation_tool],
    system_message="""你是一个自主编程Agent。收到需求后：
    1. 分析需求，拆解为子任务
    2. 逐个实现子任务
    3. 每次实现后运行测试
    4. 如果测试失败，分析错误并修复
    5. 全部完成后进行代码审查
    完成后说'CODE_COMPLETE'。"""
)

reviewer = AssistantAgent(
    name="reviewer",
    model_client=model_client,
    system_message="""你是代码审查专家。审查代码的：
    1. 正确性  2. 安全性  3. 性能  4. 可维护性
    通过后说'REVIEW_APPROVED'，否则指出具体问题。"""
)

# 选择器函数——LLM决定谁该发言
def selector_func(messages):
    last_msg = messages[-1]
    if "CODE_COMPLETE" in last_msg.content:
        return "reviewer"
    elif "REVIEW_APPROVED" in last_msg.content:
        return "user"  # 结束
    else:
        return "coder"

team = SelectorGroupChat(
    [coder, reviewer],
    model_client=model_client,
    selector_func=selector_func,
    termination_condition=TextMentionTermination("REVIEW_APPROVED") | 
                          MaxMessageTermination(30)
)

result = await team.run(task="实现一个支持增删改查的RESTful API，包含单元测试")

from claude_agent import Agent, MCPToolServer, Guardrail
from claude_agent.memory import ConversationMemory

# 定义安全护栏
def code_execution_guard(tool_name: str, tool_args: dict) -> bool:
    """阻止危险的代码执行操作"""
    dangerous_patterns = ["rm -rf", "DROP TABLE", "format c:", "__import__('os').system"]
    if tool_name == "execute_python":
        code = tool_args.get("code", "")
        for pattern in dangerous_patterns:
            if pattern in code:
                return False  # 阻止执行
    return True

def output_quality_guard(output: str) -> str:
    """确保输出质量——过滤幻觉内容"""
    # 检查是否有"我没有确切数据但..."
    hedging_patterns = ["我不确定", "可能", "大概", "据我估计"]
    for pattern in hedging_patterns:
        if pattern in output:
            return output + "\n\n⚠️ 以上内容包含不确定性陈述，请验证关键数据。"
    return output

agent = Agent(
    name="研究助手",
    instructions="""你是一个严谨的研究助手。
    原则：
    1. 只陈述有来源的事实
    2. 不确定的内容明确标注
    3. 每次工具调用前评估必要性
    4. Token消耗超过预算80%时主动总结""",
    model="claude-sonnet-4-6",
    tools=[
        MCPToolServer("web-search", "http://localhost:3001"),
        MCPToolServer("code-interpreter", "http://localhost:3002"),
        MCPToolServer("database", "http://localhost:3003"),
    ],
    guardrails=[
        Guardrail("pre_tool", code_execution_guard),
        Guardrail("post_output", output_quality_guard),
    ],
    memory=ConversationMemory(max_turns=50),
    max_iterations=30,
    token_budget=100_000
)

# 流式执行——实时观察Agent思考过程
async for event in agent.run_stream("分析全球AI芯片供应链的脆弱性"):
    if event.type == "thinking":
        print(f"💭 思考: {event.content}")
    elif event.type == "tool_call":
        print(f"🔧 调用工具: {event.tool_name}({event.args})")
    elif event.type == "tool_result":
        print(f"📋 工具结果: {event.result[:200]}...")
    elif event.type == "output":
        print(f"✅ 最终输出: {event.content}")

from google.adk import Agent, PlanningConfig, ExecutionConfig
from google.adk.tools import google_search, code_execution, bigquery_tool

research_agent = Agent(
    name="深度研究Agent",
    model="gemini-2.5-pro",
    instruction="""你是一个自主研究Agent。你的工作流程：

    Planning阶段（每5步执行一次）：
    - 回顾已完成的研究步骤
    - 评估当前信息是否足够回答问题
    - 决定下一步研究方向
    - 如果发现死胡同，调整策略

    Execution阶段（每步执行）：
    - 根据当前计划执行一个具体操作
    - 搜索信息、分析数据、总结发现
    - 将结果记录到共享状态中""",

    tools=[google_search, code_execution, bigquery_tool],

    planning_config=PlanningConfig(
        model="gemini-2.5-pro",  # 规划用强模型
        frequency=5,              # 每5步重新规划
        max_planning_rounds=10,
        strategy="adaptive"       # 自适应：根据进展动态调整
    ),

    execution_config=ExecutionConfig(
        model="gemini-2.5-flash",  # 执行用轻量模型
        max_steps=50,
        timeout_per_step=30,
        retry_on_failure=True,
        max_retries=3
    ),

    session_config={
        "save_intermediate": True,
        "checkpoint_interval": 5,  # 每5步保存检查点
    }
)

# 执行长时间自主研究任务
result = await research_agent.run(
    "分析2020-2026年全球电动汽车市场的结构性变化，"
    "识别三大转折点及其驱动因素，预测2027-2030年趋势"
)

# 从检查点恢复——如果中途失败
result = await research_agent.resume_from_checkpoint(
    checkpoint_id="ev-research-20260519",
    thread_id="research-001"
)

graph LR
    subgraph 闭环质量指标
        A[目标理解准确率] --> B[首次工具选择正确率]
        B --> C[失败感知率]
        C --> D[策略调整有效率]
        D --> E[最终任务完成率]
    end

quadrantChart
    title 自主智能体框架选型矩阵
    x-axis "低自主性" --> "高自主性"
    y-axis "低生产就绪" --> "高生产就绪"
    quadrant-1 "理想区"
    quadrant-2 "潜力区"
    quadrant-3 "实验区"
    quadrant-4 "稳定区"
    "OpenAI Agents": [0.75, 0.80]
    "LangGraph": [0.85, 0.92]
    "CrewAI": [0.70, 0.60]
    "AutoGen 0.4": [0.80, 0.65]
    "Claude Agent": [0.88, 0.78]
    "Google ADK": [0.72, 0.55]