AI Agent 2025：智能体框架的战国时代

from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

# 1. 定义工具
def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：{query} 的相关信息..."

def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

# 2. 构建 Agent 图
class AgentState(MessagesState):
    pass

graph = StateGraph(AgentState)

# 3. 添加节点
model = ChatOpenAI(model="gpt-4o").bind_tools([search_web, calculate])
tool_node = ToolNode([search_web, calculate])

def call_model(state: AgentState):
    return {"messages": [model.invoke(state["messages"])]}

graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)

# 4. 定义边：判断是否继续调用工具
def should_continue(state: AgentState):
    last_msg = state["messages"][-1]
    return "tools" if last_msg.tool_calls else "__end__"

graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

# 5. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [("human", "2025年Go发布了哪个版本？有什么新特性？")]})

from crewai import Agent, Task, Crew, Process

# 1. 定义 Agent 角色
researcher = Agent(
    role="技术研究员",
    goal="搜索并总结最新的技术趋势",
    backstory="你是一名资深技术分析师，擅长从海量信息中提取关键洞察。",
    tools=[search_web],
    verbose=True,
)

writer = Agent(
    role="技术写手",
    goal="将研究结果写成易懂的技术文章",
    backstory="你是一名技术博客作者，擅长将复杂技术概念讲清楚。",
    verbose=True,
)

# 2. 定义任务
research_task = Task(
    description="搜索 2025 年 Go 语言的最新进展",
    expected_output="一份包含关键特性、版本号和发布日期的技术报告",
    agent=researcher,
)

write_task = Task(
    description="根据研究报告撰写一篇技术博客",
    expected_output="一篇结构清晰、适合开发者阅读的博客文章",
    agent=writer,
)

# 3. 组建团队并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 或 Process.hierarchical
)

result = crew.kickoff()
print(result)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpToolAdapter

# 1. 注册 MCP 工具
mcp_tools = McpToolAdapter.from_server("weather-mcp")

# 2. 创建 Agent
weather_agent = AssistantAgent(
    name="weather_agent",
    model_client=OpenAIChatCompletionClient(model="gpt-4o"),
    tools=mcp_tools,
    system_message="你是天气查询助手，使用 MCP 工具获取实时天气。",
)

planner_agent = AssistantAgent(
    name="planner_agent",
    model_client=OpenAIChatCompletionClient(model="gpt-4o"),
    system_message="你是计划助手，根据天气信息给出出行建议。",
)

# 3. 组成团队
team = RoundRobinGroupChat([weather_agent, planner_agent])

# 4. 运行
stream = team.run_stream(task="明天北京适合出门吗？")
async for message in stream:
    print(message)

# 一个 MCP 服务器的简单示例
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("code-analyzer")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="analyze_code",
            description="分析代码中的问题",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {"type": "string"},
                    "language": {"type": "string"},
                },
            },
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_code":
        # 执行代码分析
        return [TextContent(type="text", text="分析结果...")]

import asyncio
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from mcp import ClientSession, StdioServerParameters

# 1. 连接 MCP 服务器
async def setup_mcp_tools():
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "mcp_server_code_review"],
    )
    async with ClientSession(server_params) as session:
        tools = await session.list_tools()
        return [convert_mcp_to_langchain(t) for t in tools]

# 2. 构建审查 Agent
def build_review_agent():
    tools = asyncio.run(setup_mcp_tools())
    tool_node = ToolNode(tools)
    
    model = ChatOpenAI(model="gpt-4o").bind_tools(tools)
    
    graph = StateGraph(MessagesState)
    graph.add_node("agent", lambda s: {"messages": [model.invoke(s["messages"])]})
    graph.add_node("tools", tool_node)
    graph.set_entry_point("agent")
    graph.add_conditional_edges(
        "agent",
        lambda s: "tools" if s["messages"][-1].tool_calls else "__end__",
    )
    graph.add_edge("tools", "agent")
    
    return graph.compile()

# 3. 审查代码
agent = build_review_agent()
result = agent.invoke({
    "messages": [("human", """审查下面这段 Go 代码：\n\nfunc main() {\n    for i := 0; i < 10; i++ {\n        go func() {\n            fmt.Println(i)\n        }()\n    }\n}""")]
})