Agent 框架

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 实际项目中调用搜索 API
    return f"搜索 '{query}' 的结果：这里是搜索到的相关信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{e}"

# 3. 配置模型
tools = [search_web, calculator]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 4. 定义节点
def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

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

# 6. 定义边
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent")

# 7. 编译并运行
app = graph.compile()

# 使用 Agent
result = app.invoke({
    "messages": [("user", "帮我计算 (125 * 37) + 892 的结果")]
})
print(result["messages"][-1].content)

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

# 将索引查询引擎包装为工具
query_tool = QueryEngineTool.from_defaults(
    query_engine=index.as_query_engine(),
    name="knowledge_base",
    description="查询内部知识库获取公司文档信息"
)

# 自定义函数工具
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件通知"""
    # 实际邮件发送逻辑
    return f"邮件已发送至 {to}"

email_tool = FunctionTool.from_defaults(fn=send_email)

# 创建 ReAct Agent
agent = ReActAgent.from_tools(
    tools=[query_tool, email_tool],
    llm=OpenAI(model="gpt-4o"),
    verbose=True
)

response = agent.chat("查询最新的项目报告并发送给团队负责人")

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 创建多个专业 Agent
coder = AssistantAgent(
    name="Coder",
    system_message="你是一位 Python 专家，负责编写高质量代码。",
    llm_config={"model": "gpt-4o"}
)

reviewer = AssistantAgent(
    name="Reviewer",
    system_message="你是一位代码审查专家，负责审查代码质量和安全性。",
    llm_config={"model": "gpt-4o"}
)

# 用户代理（可执行代码）
user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace"}
)

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

manager = GroupChatManager(groupchat=group_chat)

# 发起任务
user_proxy.initiate_chat(
    manager,
    message="编写一个 Python 脚本来分析 CSV 文件中的销售数据并生成图表"
)

from crewai import Agent, Task, Crew, Process

# 定义 Agent 角色
researcher = Agent(
    role="高级研究员",
    goal="深入研究给定主题并提供详尽的分析报告",
    backstory="你是一位经验丰富的技术研究员，擅长信息收集和分析",
    tools=[search_tool],
    verbose=True
)

writer = Agent(
    role="技术写作专家",
    goal="将研究结果转化为通俗易懂的技术文章",
    backstory="你是一位资深技术作家，擅长用简洁的语言解释复杂概念",
    verbose=True
)

# 定义任务
research_task = Task(
    description="研究 AI Agent 在企业落地中的最佳实践",
    expected_output="一份包含关键发现、案例分析和趋势预测的研究报告",
    agent=researcher
)

writing_task = Task(
    description="基于研究报告撰写一篇技术博客文章",
    expected_output="一篇 2000 字左右的技术博客，结构清晰，案例丰富",
    agent=writer
)

# 组建团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

result = crew.kickoff()