LLM 智能体与 Agentic AI 完全指南：从 ReAct、多智能体编排到 MCP

Thought: 分析当前状况，决定下一步行动
Action: 以 tool_name(arguments) 的形式调用工具
Observation: 接收工具执行结果
...重复...
Final Answer: 得出最终答案

from langchain.agents import create_react_agent
from langchain_anthropic import ChatAnthropic
from langchain.tools import DuckDuckGoSearchRun, PythonREPLTool
from langchain import hub

llm = ChatAnthropic(model="claude-opus-4-5", temperature=0)
tools = [DuckDuckGoSearchRun(), PythonREPLTool()]

# 加载 ReAct 提示词模板
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)

from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "搜索 2026 年目前最受欢迎的 3 个 AI 智能体框架，并制作一张比较表"
})

cot_prompt = """
请一步一步解决问题：

问题：{problem}

求解过程：
1. 首先整理已给出的信息。
2. 执行所需的计算/推理。
3. 验证中间结果。
4. 得出最终答案。
"""

from langchain_experimental.tot.base import ToTChain
from langchain_experimental.tot.thought_generation import ProposePromptStrategy

tot_chain = ToTChain.from_llm(
    llm=llm,
    checker=checker,
    k=3,           # 每一层要生成的分支数
    c=4,           # 评估深度
    verbose=True
)

from mem0 import Memory

# 初始化 mem0（用 Qdrant 作为向量数据库）
config = {
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "collection_name": "agent_memory",
            "host": "localhost",
            "port": 6333,
        }
    },
    "llm": {
        "provider": "anthropic",
        "config": {
            "model": "claude-opus-4-5",
            "temperature": 0,
        }
    }
}

memory = Memory.from_config(config)
user_id = "user_123"

# 存储记忆
memory.add(
    messages=[
        {"role": "user", "content": "我主要用 Python，喜欢 FastAPI"},
        {"role": "assistant", "content": "明白了！我会记住你偏好 Python/FastAPI。"}
    ],
    user_id=user_id
)

# 检索并使用记忆
relevant_memories = memory.search(
    query="用户偏好的语言是什么？",
    user_id=user_id
)

context = "\n".join([m["memory"] for m in relevant_memories])
print(f"相关记忆：{context}")

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from datetime import datetime

class EpisodicMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.store = Chroma(
            collection_name="episodes",
            embedding_function=self.embeddings,
            persist_directory="./episodic_memory"
        )

    def store_episode(self, content: str, metadata: dict = None):
        """将对话/任务片段存储到记忆中"""
        metadata = metadata or {}
        metadata["timestamp"] = datetime.now().isoformat()
        self.store.add_texts([content], metadatas=[metadata])

    def recall(self, query: str, k: int = 3):
        """检索相关片段"""
        docs = self.store.similarity_search(query, k=k)
        return [doc.page_content for doc in docs]

# 使用示例
memory = EpisodicMemory()
memory.store_episode(
    "用户询问了 FastAPI 项目结构相关的问题，并得到了成功的回答",
    {"task_type": "coding", "success": True}
)

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import httpx
from typing import Optional

class WebSearchInput(BaseModel):
    query: str = Field(description="要搜索的查询")
    max_results: int = Field(default=5, description="返回结果的数量")

class TavilySearchTool(BaseTool):
    name: str = "web_search"
    description: str = "在网页上搜索最新信息。需要实时信息时使用。"
    args_schema: type[BaseModel] = WebSearchInput
    api_key: str = ""

    def _run(self, query: str, max_results: int = 5) -> str:
        url = "https://api.tavily.com/search"
        payload = {
            "api_key": self.api_key,
            "query": query,
            "max_results": max_results,
            "include_answer": True,
        }
        response = httpx.post(url, json=payload)
        data = response.json()

        results = []
        if data.get("answer"):
            results.append(f"摘要: {data['answer']}\n")
        for r in data.get("results", []):
            results.append(f"- {r['title']}: {r['content'][:200]}...")
        return "\n".join(results)

    async def _arun(self, query: str, max_results: int = 5) -> str:
        # 异步版本
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.tavily.com/search",
                json={"api_key": self.api_key, "query": query, "max_results": max_results}
            )
        return self._process_response(response.json())

# MCP 服务器实现示例（Python SDK）
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("my-tool-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="查询指定城市的当前天气",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_weather":
        city = arguments["city"]
        # 实际调用天气 API
        weather_data = await fetch_weather(city)
        return [types.TextContent(type="text", text=str(weather_data))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from typing import TypedDict, Annotated, Sequence
import operator

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence, operator.add]
    tool_calls: list
    iteration_count: int

# 2. 设置 LLM 与工具
llm = ChatAnthropic(model="claude-opus-4-5")
tools = [WebSearchTool(), PythonREPLTool()]
llm_with_tools = llm.bind_tools(tools)

# 3. 定义节点
def call_model(state: AgentState) -> AgentState:
    """LLM 调用节点"""
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration_count": state["iteration_count"] + 1
    }

def call_tools(state: AgentState) -> AgentState:
    """工具执行节点"""
    last_message = state["messages"][-1]
    tool_results = []

    for tool_call in last_message.tool_calls:
        tool = next(t for t in tools if t.name == tool_call["name"])
        result = tool.invoke(tool_call["args"])
        tool_results.append(
            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
        )
    return {"messages": tool_results}

# 4. 路由函数（条件边）
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    # 有工具调用则继续，没有则结束
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        if state["iteration_count"] < 10:  # 防止无限循环
            return "tools"
    return "end"

# 5. 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)

graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")  # 工具执行后回到智能体

# 6. 添加记忆检查点
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 执行（用 thread_id 管理对话会话）
config = {"configurable": {"thread_id": "session_001"}}
result = app.invoke(
    {"messages": [HumanMessage(content="搜索并总结 2026 年 AI 智能体的趋势")], "iteration_count": 0},
    config=config
)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileWriterTool

# 工具设置
search_tool = SerperDevTool()
file_writer = FileWriterTool()

# 智能体定义
researcher = Agent(
    role="AI 研究员",
    goal="深入调研最新的 AI 智能体技术趋势",
    backstory="""你是 AI 领域的专业研究员。
    你会分析最新的论文、博客和 GitHub 仓库，提炼出核心洞见。""",
    tools=[search_tool],
    llm="claude-opus-4-5",
    verbose=True
)

writer = Agent(
    role="技术作家",
    goal="把调研结果写成易读的技术报告",
    backstory="""你是一名擅长把复杂 AI 概念讲清楚的专业作家。""",
    tools=[file_writer],
    llm="claude-opus-4-5",
    verbose=True
)

# 任务定义
research_task = Task(
    description="调研 2026 年 LLM 智能体的 5 大趋势。每个趋势都要包含具体案例和影响。",
    expected_output="5 个趋势的详细分析（每个至少 500 字）",
    agent=researcher
)

writing_task = Task(
    description="基于调研结果撰写一篇技术博客文章。",
    expected_output="一篇 2000 字左右、Markdown 格式的技术博客文章",
    agent=writer,
    output_file="ai_trends_2026.md"
)

# 执行 Crew（顺序流程）
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()

import autogen

config_list = [{"model": "claude-opus-4-5", "api_key": "YOUR_KEY"}]

# 编排智能体
orchestrator = autogen.AssistantAgent(
    name="Orchestrator",
    system_message="""你是负责统筹团队的编排者。
    你会分析任务，并将其委派给合适的专家智能体。
    最后整合所有结果，生成最终答案。""",
    llm_config={"config_list": config_list}
)

# 代码执行智能体
coder = autogen.AssistantAgent(
    name="Coder",
    system_message="你是一名专门编写并执行 Python 代码的专家。",
    llm_config={"config_list": config_list, "functions": [...]}
)

# 用户代理（负责代码执行）
user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

# 执行群聊
groupchat = autogen.GroupChat(
    agents=[orchestrator, coder, user_proxy],
    messages=[],
    max_round=12
)
manager = autogen.GroupChatManager(groupchat=groupchat)
user_proxy.initiate_chat(manager, message="写一段数据可视化的代码")

import anthropic
import json

client = anthropic.Anthropic()

# 工具定义
tools = [
    {
        "name": "get_stock_price",
        "description": "查询指定股票的当前股价和涨跌幅",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "股票代码（例：AAPL、MSFT）"
                },
                "currency": {
                    "type": "string",
                    "enum": ["USD", "KRW"],
                    "description": "显示货币"
                }
            },
            "required": ["symbol"]
        }
    }
]

def process_tool_call(tool_name: str, tool_input: dict) -> str:
    """工具执行逻辑"""
    if tool_name == "get_stock_price":
        # 实际 API 调用（示例）
        return json.dumps({
            "symbol": tool_input["symbol"],
            "price": 185.92,
            "change_percent": "+2.3%",
            "currency": tool_input.get("currency", "USD")
        })

# 智能体循环
messages = [{"role": "user", "content": "告诉我 Apple 的股价"}]

while True:
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        tools=tools,
        messages=messages
    )

    messages.append({"role": "assistant", "content": response.content})

    if response.stop_reason == "end_turn":
        # 提取最终文本回答
        final_text = next(
            block.text for block in response.content
            if hasattr(block, "text")
        )
        print(f"最终答案: {final_text}")
        break

    if response.stop_reason == "tool_use":
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = process_tool_call(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        messages.append({"role": "user", "content": tool_results})

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class AgentTrajectory:
    task: str
    steps: List[dict]  # {"thought": ..., "action": ..., "observation": ...}
    final_answer: str
    success: bool
    total_tokens: int

def evaluate_trajectory(trajectory: AgentTrajectory) -> dict:
    """基于路径的智能体评估"""
    metrics = {
        "task_success": trajectory.success,
        "efficiency": calculate_efficiency(trajectory.steps),
        "redundant_steps": count_redundant_steps(trajectory.steps),
        "error_recovery": check_error_recovery(trajectory.steps),
        "tool_usage_appropriateness": evaluate_tool_usage(trajectory.steps),
        "cost_efficiency": 1000 / trajectory.total_tokens  # 单位 token 的效率
    }
    return metrics

def count_redundant_steps(steps: List[dict]) -> int:
    """不必要的重复工具调用次数"""
    seen_actions = set()
    redundant = 0
    for step in steps:
        action_key = f"{step.get('action_type')}:{step.get('action_input')}"
        if action_key in seen_actions:
            redundant += 1
        seen_actions.add(action_key)
    return redundant

import anthropic
import base64
from PIL import ImageGrab

def take_screenshot() -> str:
    """截取屏幕并进行 base64 编码"""
    screenshot = ImageGrab.grab()
    screenshot.save("/tmp/screenshot.png")
    with open("/tmp/screenshot.png", "rb") as f:
        return base64.b64encode(f.read()).decode()

client = anthropic.Anthropic()

# Computer-use 智能体
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=4096,
    tools=[
        {"type": "computer_20241022", "name": "computer", "display_width_px": 1920, "display_height_px": 1080},
        {"type": "text_editor_20241022", "name": "str_replace_editor"},
        {"type": "bash_20241022", "name": "bash"}
    ],
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "打开浏览器，查看 GitHub 上最新的热门仓库"},
            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": take_screenshot()}}
        ]
    }]
)

from openai import OpenAI
import time

client = OpenAI()

# 创建 Assistant（包含工具 + 知识库）
assistant = client.beta.assistants.create(
    name="AI 技术分析师",
    instructions="你是一名 AI/ML 技术专家。请分析最新的论文和技术文档，提供洞见。",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},   # 基于文件的 RAG
        {"type": "code_interpreter"}  # 代码执行
    ]
)

# 上传文件并创建向量存储
vector_store = client.beta.vector_stores.create(name="AI 论文库")
with open("ai_papers_2026.pdf", "rb") as f:
    client.beta.vector_stores.file_batches.upload_and_poll(
        vector_store_id=vector_store.id,
        files=[("ai_papers_2026.pdf", f)]
    )

# 创建 Thread 并添加消息
thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="从已上传的论文中分析 Agentic AI 的主要局限性"
)

# 执行 Run 并等待结果
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

while run.status in ["queued", "in_progress"]:
    run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
    time.sleep(1)

# 输出结果
messages = client.beta.threads.messages.list(thread_id=thread.id)
print(messages.data[0].content[0].text.value)