from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain.messages import SystemMessage, HumanMessage, ToolMessage

# 定义工具
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

tools = [multiply, add]
tools_by_name = {tool.name: tool for tool in tools}

# 定义节点
def llm_call(state: MessagesState):
    """LLM 决定是否调用工具"""
    return {
        "messages": [
            llm_with_tools.invoke(
                [SystemMessage(content="You are a helpful arithmetic assistant.")]
                + state["messages"]
            )
        ]
    }

def tool_node(state: dict):
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=str(observation), tool_call_id=tool_call["id"]))
    return {"messages": result}

# 定义条件边路由
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

# 构建并编译图
builder = StateGraph(MessagesState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
builder.add_edge("tool_node", "llm_call")

agent = builder.compile()

# 运行
result = agent.invoke({"messages": [HumanMessage(content="Add 3 and 4, then multiply by 5.")]})

from crewai import Agent, Task, Crew, Process

# 定义 Agent
researcher = Agent(
    role='高级 AI 研究员',
    goal='发现 AI Agent 领域的最新发展趋势',
    backstory='你是一位经验丰富的研究员，擅长发现前沿技术的最新动态。',
    verbose=True,
    memory=True,
)

writer = Agent(
    role='技术报告撰写专家',
    goal='将研究发现转化为清晰、结构化的报告',
    backstory='你是一位技术写作专家，擅长将复杂信息转化为易读的报告。',
    verbose=True,
    memory=True,
)

# 定义 Task
research_task = Task(
    description='对 {topic} 进行全面调研，收集最新的发展趋势。',
    expected_output='包含 10 个要点的详细研究发现列表',
    agent=researcher,
)

writing_task = Task(
    description='根据研究发现撰写一份完整的技术报告。',
    expected_output='完整的 Markdown 格式报告',
    agent=writer,
    context=[research_task],
    output_file='report.md',
)

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

result = crew.kickoff(inputs={'topic': '多智能体协作系统'})

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
import asyncio

# 构建 RAG 索引
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

# 定义工具
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

async def search_documents(query: str) -> str:
    """Search documents for answers."""
    response = await query_engine.aquery(query)
    return str(response)

# 创建智能体
agent = FunctionAgent(
    tools=[multiply, search_documents],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="You are a helpful assistant that can calculate and search documents.",
)

# 运行
async def main():
    response = await agent.run("What did the author do in college? Also, what's 7 * 8?")
    print(response)

asyncio.run(main())

import asyncio
from agents import Agent, Runner, function_tool

# 定义工具
@function_tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"The weather in {city} is sunny."

# 定义专业 Agent
billing_agent = Agent(
    name="Billing Agent",
    instructions="你是账单问题专家。",
)

refund_agent = Agent(
    name="Refund Agent",
    instructions="你是退款问题专家。",
)

# 定义分流 Agent
triage_agent = Agent(
    name="Triage Agent",
    instructions="根据用户问题路由到正确的专业 Agent：账单 -> Billing Agent；退款 -> Refund Agent。",
    handoffs=[billing_agent, refund_agent],
    tools=[get_weather],
)

# 运行
async def main():
    result = await Runner.run(
        triage_agent,
        "我的订阅被扣了两次费用，请帮我处理。",
    )
    print(f"最终回答: {result.final_output}")
    print(f"处理 Agent: {result.last_agent.name}")

asyncio.run(main())

import asyncio
from google.adk.agents import Agent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.tools import google_search

# 定义天气 Agent
weather_agent = Agent(
    name="weather_assistant",
    model="gemini-2.5-flash",
    instruction="你是一个天气查询助手。使用 Google 搜索查找最新天气信息。",
    tools=[google_search],
)

# 定义翻译 Agent
translate_agent = Agent(
    name="translate_assistant",
    model="gemini-2.5-flash",
    instruction="你是一个翻译助手，将内容翻译成中文。",
)

# 组合成顺序工作流
pipeline = SequentialAgent(
    name="WeatherPipeline",
    sub_agents=[weather_agent, translate_agent],
)

# 运行
session_service = InMemorySessionService()
runner = Runner(agent=pipeline, app_name="weather_app", session_service=session_service)

async def run_agent(query: str):
    session = session_service.create_session(
        app_name="weather_app", user_id="user_1", session_id="session_1"
    )
    content = types.Content(role='user', parts=[types.Part(text=query)])
    async for event in runner.run_async(
        user_id="user_1", session_id="session_1", new_message=content
    ):
        if event.is_final_response() and event.content and event.content.parts:
            print(f"Agent 回复: {event.content.parts[0].text.strip()}")

asyncio.run(run_agent("What's the weather in Tokyo today?"))