Agent 框架深度对比：LangChain / LlamaIndex / CrewAI / AutoGen 能力矩阵与实战

from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# LCEL 管道：输入 → retriever → prompt → LLM → 输出
model = ChatOpenAI(model="gpt-4o")

# 基础链
chain = (
    RunnablePassthrough.assign(context=lambda x: retriever.invoke(x["question"]))
    | ChatPromptTemplate.from_template("""
Answer based on context:
{context}

Question: {question}
""")
    | model
    | StrOutputParser()
)

# 并行执行
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain,
    sentiment=sentiment_chain,
)

from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI

# 定义工具
tools = [
    Tool(
        name="search",
        func=lambda q: search_engine(q),
        description="Search the web for current information",
    ),
    Tool(
        name="calculator",
        func=lambda expr: eval(expr),  # 生产环境用 safe_eval
        description="Evaluate mathematical expressions",
    ),
    Tool(
        name="weather",
        func=lambda city: f"Weather in {city}: 22°C, sunny",
        description="Get current weather for a city",
    ),
]

# 使用标准 ReAct prompt
prompt = hub.pull("hwchase17/react")

llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # 打印思考过程
    max_iterations=10,      # 防止无限循环
    handle_parsing_errors=True,
)

# 执行
result = agent_executor.invoke({
    "input": "What's the weather in Tokyo and what is 234 * 567?"
})
print(result["output"])

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain.tools import StructuredTool

class SearchInput(BaseModel):
    query: str = Field(description="Search query")
    top_k: int = Field(default=5, description="Number of results")

search_tool = StructuredTool.from_function(
    func=lambda query, top_k: search_api(query, top_k),
    name="smart_search",
    description="Search with configurable result count",
    args_schema=SearchInput,
)

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

# 定义节点
def agent_node(state: AgentState):
    # Agent 决定下一步
    response = llm.invoke(state["messages"])
    return {"messages": [response], "next_step": "tool" if "Action:" in response.content else "end"}

def tool_node(state: AgentState):
    # 执行工具
    result = execute_tool(state["messages"][-1].content)
    return {"messages": [f"Observation: {result}"]}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tool", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    lambda state: state["next_step"],
    {"tool": "tool", "end": END},
)
workflow.add_edge("tool", "agent")

graph = workflow.compile()

# 运行
result = graph.invoke({"messages": ["What's the weather?"]})

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

# 创建检索工具
query_engine = index.as_query_engine(similarity_top_k=3)
retrieval_tool = FunctionTool.from_defaults(
    fn=lambda query: str(query_engine.query(query)),
    name="knowledge_retriever",
    description="Retrieve information from the knowledge base",
)

# 创建 LlamaIndex ReAct Agent
llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools(
    tools=[retrieval_tool, calculator_tool],
    llm=llm,
    verbose=True,
    max_iterations=10,
)

response = agent.chat("根据知识库，解释 RAG 架构，并计算 100 个 512 维向量占多少内存？")
print(response)

from llama_index.agent.openai import OpenAIAgent

# 直接使用 OpenAI Function Calling，无需 ReAct 循环
agent = OpenAIAgent.from_tools(tools, verbose=True)
response = agent.chat("What's 15% of 2340?")

from crewai import Agent, Task, Crew
from crewai.tools import tool
from langchain_openai import ChatOpenAI

@tool
def research_topic(topic: str) -> str:
    """Research a topic and return summary."""
    # 连接到搜索引擎或 RAG 系统
    return f"Research results for {topic}: ..."

@tool
def write_content(outline: str) -> str:
    """Write content based on outline."""
    return f"Content based on outline: {outline}"

# 定义角色
researcher = Agent(
    role="Research Analyst",
    goal="Find comprehensive information on topics",
    backstory="You are an expert researcher with 10 years of experience.",
    tools=[research_topic],
    llm=ChatOpenAI(model="gpt-4o-mini"),
    verbose=True,
)

writer = Agent(
    role="Content Writer",
    goal="Create engaging content from research",
    backstory="You are a professional writer with deep technical knowledge.",
    tools=[write_content],
    llm=ChatOpenAI(model="gpt-4o"),
    verbose=True,
)

# 定义任务
research_task = Task(
    description="Research the latest developments in quantum computing.",
    agent=researcher,
    expected_output="A comprehensive summary of recent quantum computing advances.",
)

writing_task = Task(
    description="Write a blog post based on the research findings.",
    agent=writer,
    expected_output="A 1000-word blog post in markdown format.",
    context=[research_task],  # 依赖前一个任务的输出
)

# 创建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process="sequential",  # sequential / hierarchical / parallel
    verbose=2,
)

result = crew.kickoff()
print(result)

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 配置 LLM
config_list = [
    {
        "model": "gpt-4o",
        "api_key": os.getenv("OPENAI_API_KEY"),
    }
]

# 创建 Agent
coder = AssistantAgent(
    name="Coder",
    llm_config={"config_list": config_list},
    system_message="You are a Python expert. Write clean, well-documented code.",
)

reviewer = AssistantAgent(
    name="Reviewer",
    llm_config={"config_list": config_list},
    system_message="You review code for bugs, style issues, and best practices.",
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

# 群聊
group_chat = GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=12,
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"config_list": config_list},
)

# 启动对话
user_proxy.initiate_chat(
    manager,
    message="Write a FastAPI endpoint that calculates Fibonacci numbers with caching.",
)

# 嵌套对话（Agent 之间子对话）
from autogen import initiate_chats

chat_results = initiate_chats([
    {"sender": user_proxy, "recipient": coder, "message": "Write the function", "summary_method": "reflection_with_llm"},
    {"sender": reviewer, "recipient": coder, "message": "Review the code", "summary_method": "last_msg"},
])

# 代码执行（沙箱）
coder_with_exec = AssistantAgent(
    name="CoderExec",
    llm_config={"config_list": config_list},
    code_execution_config={
        "work_dir": "sandbox",
        "use_docker": True,  # Docker 沙箱执行
    },
)

class ToolRegistry:
    """工具注册中心。"""
    def __init__(self):
        self.tools: dict[str, callable] = {}
        self.schemas: list[dict] = []

    def register(self, name: str, description: str, schema: dict, fn: callable):
        self.tools[name] = fn
        self.schemas.append({
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": schema,
            },
        })

    def execute(self, name: str, arguments: dict) -> str:
        if name not in self.tools:
            return f"Error: Tool '{name}' not found"
        try:
            result = self.tools[name](**arguments)
            return str(result)
        except Exception as e:
            return f"Error executing {name}: {e}"

# 注册工具
registry = ToolRegistry()
registry.register(
    name="get_weather",
    description="Get weather for a city",
    schema={
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "unit": {"type": "string", "enum": ["C", "F"]},
        },
        "required": ["city"],
    },
    fn=lambda city, unit="C": f"{city}: 22°{unit}",
)

# MCP 客户端（概念性示例，使用 mcp SDK）
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def mcp_client_demo():
    # 启动 MCP 服务器（如文件系统服务器）
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出可用工具
            tools = await session.list_tools()
            for tool in tools:
                print(f"Tool: {tool.name} - {tool.description}")

            # 调用工具
            result = await session.call_tool(
                "read_file",
                {"path": "/path/to/files/document.txt"}
            )
            print(result)

from typing import List
import json

class MemoryManager:
    """分层记忆管理。"""
    def __init__(self, vector_store, max_short_term: int = 10):
        self.short_term: List[dict] = []       # 最近 N 轮对话
        self.vector_store = vector_store        # 长期记忆向量库
        self.entity_memory: dict = {}           # 实体记忆
        self.max_short_term = max_short_term

    def add_message(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        if len(self.short_term) > self.max_short_term:
            # 移出的消息压缩存入长期记忆
            old = self.short_term.pop(0)
            self._to_long_term(old)

    def _to_long_term(self, message: dict):
        """将消息编码为 embedding 存入向量库。"""
        embedding = embedder.encode(message["content"])
        self.vector_store.upsert(
            [message["content"]],
            [embedding],
            [{"role": message["role"], "timestamp": time.time()}],
        )

    def get_relevant_memories(self, query: str, top_k: int = 3) -> List[str]:
        """检索与当前查询相关的长期记忆。"""
        query_emb = embedder.encode(query)
        results = self.vector_store.search(query_emb, top_k)
        return [r["text"] for r in results]

    def build_context(self, query: str, system_prompt: str) -> List[dict]:
        """构建完整的 LLM 上下文。"""
        messages = [{"role": "system", "content": system_prompt}]

        # 添加相关长期记忆
        memories = self.get_relevant_memories(query)
        if memories:
            memory_text = "\n".join(f"- {m}" for m in memories)
            messages.append({
                "role": "system",
                "content": f"Relevant past information:\n{memory_text}",
            })

        # 添加短期记忆
        messages.extend(self.short_term)

        # 当前查询
        messages.append({"role": "user", "content": query})
        return messages

    def extract_entities(self, text: str):
        """抽取并更新实体记忆。"""
        # 使用 NER 或 LLM 抽取
        prompt = f"Extract entities (PERSON, ORG, LOCATION) from: {text}\nOutput JSON."
        result = llm.chat_completion([{"role": "user", "content": prompt}])
        entities = json.loads(result)
        for ent in entities:
            self.entity_memory[ent["name"]] = ent

from langchain.memory import ConversationBufferMemory, VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma

# 短期：缓冲记忆
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
)

# 长期：向量检索记忆
vectorstore = Chroma(embedding_function=embeddings)
retriever_memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(),
    memory_key="historical_context",
)

# Agent 结合两种记忆
agent = create_react_agent(
    llm, tools,
    prompt=prompt.partial(
        chat_history=lambda x: buffer_memory.load_memory_variables(x)["chat_history"],
        historical_context=lambda x: retriever_memory.load_memory_variables(x)["historical_context"],
    ),
)

from typing import AsyncIterator
import json

async def stream_react_agent(query: str, tools: dict) -> AsyncIterator[dict]:
    """流式展示 Agent 思考过程。"""
    messages = [{"role": "user", "content": query}]
    max_steps = 10

    for step in range(max_steps):
        # 流式获取 LLM 响应
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[tool["schema"] for tool in tools.values()],
            stream=True,
        )

        thought = ""
        async for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                thought += delta.content
                yield {"type": "thought", "content": delta.content}

        # 检查工具调用
        # 完整响应需要重新获取（或使用 accumulation）
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[tool["schema"] for tool in tools.values()],
        )
        msg = response.choices[0].message

        if msg.tool_calls:
            for call in msg.tool_calls:
                tool_name = call.function.name
                args = json.loads(call.function.arguments)
                yield {"type": "action", "tool": tool_name, "args": args}

                # 执行工具
                result = tools[tool_name]["fn"](**args)
                yield {"type": "observation", "content": result}

                messages.append({"role": "assistant", "content": None, "tool_calls": [call]})
                messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
        else:
            yield {"type": "final", "content": msg.content}
            break

# FastAPI SSE 端点
@app.get("/agent/stream")
async def agent_stream(query: str):
    async def event_generator():
        async for event in stream_react_agent(query, tools):
            yield f"data: {json.dumps(event)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

async def reflection_workflow(task: str):
    """写作者 + 审稿人 反射循环。"""
    writer_result = await writer_agent.run(task)
    feedback = await reviewer_agent.run(f"Review this: {writer_result}")

    if "APPROVED" in feedback:
        return writer_result

    # 迭代改进
    for iteration in range(3):
        improved = await writer_agent.run(
            f"Improve based on feedback: {feedback}\n\nOriginal: {writer_result}"
        )
        feedback = await reviewer_agent.run(f"Review: {improved}")
        if "APPROVED" in feedback:
            return improved

    return improved

class SupervisorAgent:
    """监督者 Agent 分配任务给专业 Worker。"""
    def __init__(self, workers: dict[str, Agent]):
        self.workers = workers

    async def delegate(self, task: str) -> str:
        # 让监督者决定由哪个 Worker 处理
        decision_prompt = f"""Available workers: {list(self.workers.keys())}
Task: {task}
Which worker should handle this? Respond with just the worker name."""
        worker_name = (await llm.chat_completion([
            {"role": "user", "content": decision_prompt}
        ])).strip()

        if worker_name in self.workers:
            return await self.workers[worker_name].run(task)
        return await self.workers["generalist"].run(task)

class RouterAgent:
    """根据查询类型路由到不同专业 Agent。"""
    def __init__(self):
        self.routes = {
            "code": code_agent,
            "research": research_agent,
            "creative": creative_agent,
        }

    async def route(self, query: str) -> str:
        classification = await llm.chat_completion([{
            "role": "user",
            "content": f"Classify into [code|research|creative]: {query}"
        }])

        agent = self.routes.get(classification.strip().lower(), general_agent)
        return await agent.run(query)