核心概念

@tool
def my_weather_func() -> str:
    """获取当前天气。"""
    return "It's 72°F and sunny"

async def main():
    weather_agent = Agent(
        name="WeatherAgent",
        role="Weather Assistant",
        goal="Provide timely weather updates across cities",
        instructions=["Help users with weather information"],
        tools=[my_weather_func],
        memory = AgentMemoryConfig(
            store=ConversationDaprStateMemory(
                store_name="historystore",
                session_id="some-id",
            )
        ),
    )

    response1 = await weather_agent.run("What's the weather?")
    response2 = await weather_agent.run("How about now?")

from dapr_agents.workflow.runners import AgentRunner

async def main():
    travel_planner = DurableAgent(
        name="TravelBuddy",
        role="Travel Planner",
        goal="Help users find flights and remember preferences",
        instructions=["Help users find flights and remember preferences"],
        tools=[search_flights],
        memory = AgentMemoryConfig(
            store=ConversationDaprStateMemory(
                store_name="conversationstore",
                session_id="travel-session",
            )
        )
    )

    runner = AgentRunner()

    try:
        itinerary = await runner.run(
            travel_planner,
            payload={"task": "Plan a 3-day trip to Paris"},
        )
        print(itinerary)
    finally:
        runner.shutdown(travel_planner)

# 定义数据模型
class Dog(BaseModel):
    name: str
    breed: str
    reason: str

# 初始化聊天客户端
llm = OpenAIChatClient()

# 获取结构化响应
response = llm.generate(
    messages=[UserMessage("One famous dog in history.")], response_format=Dog
)

print(json.dumps(response.model_dump(), indent=2))

@tool(args_model=GetWeatherSchema)
def get_weather(location: str) -> str:
    """根据地点获取天气信息。"""
    import random
    temperature = random.randint(60, 80)
    return f"{location}: {temperature}F."

from dapr_agents.agents.configs import AgentExecutionConfig, ToolExecutionMode

travel_planner = DurableAgent(
    name="TravelBuddy",
    ...
    execution=AgentExecutionConfig(
        max_iterations=10,
        tool_execution_mode=ToolExecutionMode.SEQUENTIAL,
    ),
)

client = MCPClient()
await client.connect_sse("local", url="http://localhost:8000/sse")

# 将 MCP 工具转换为 AgentTool 列表
tools = client.get_all_tools()

from dapr_agents import Agent, DurableAgent
from dapr_agents.agents.configs import AgentMemoryConfig
from dapr_agents.memory import (
    ConversationDaprStateMemory,
    ConversationListMemory,
    ConversationVectorMemory,
)

# 1. ConversationListMemory（简单内存）- 默认
memory_list = ConversationListMemory()

# 2. ConversationVectorMemory（向量存储）
memory_vector = ConversationVectorMemory(
    vector_store=your_vector_store_instance,
    distance_metric="cosine",
)

# 3. 通过 AgentMemoryConfig 使用 ConversationDaprStateMemory（Dapr 状态存储）
durable_memory = AgentMemoryConfig(
    store=ConversationDaprStateMemory(
        store_name="historystore",  # Dapr 组件名称
        session_id="my-session",
    )
)

# 与普通 Agent 一起使用（直接传入 memory 实例）
agent = Agent(
    name="MyAgent",
    role="Assistant",
    memory=memory_list,
)

# 与 DurableAgent 一起使用（传入 AgentMemoryConfig）
travel_planner = DurableAgent(
    name="TravelBuddy",
    memory=durable_memory,
    # ... 其他配置 ...
)

from dapr_agents.storage.vectorstores import RedisVectorStore
from dapr_agents.document.embedder.openai import OpenAIEmbedder
from dapr_agents.memory import ConversationVectorMemory

vector_store = RedisVectorStore(
    url="redis://localhost:6379",
    index_name="my_agent",
    embedding_function=OpenAIEmbedder(),
    embedding_dimensions=1536,
)

memory = ConversationVectorMemory(
    vector_store=vector_store,
    distance_metric="cosine",
)

from dapr_agents.tool.workflow import agent_to_tool

# 将独立 agent 作为一次工具调用来调用
aragorn_tool = agent_to_tool(
    "aragorn",
    description="Military Strategy. Goal: Lead the forces of Gondor.",
    target_app_id="aragorn-app",
)
# 在 DurableAgent 中把一个 agent 当作工具使用
frodo = DurableAgent(
    name="frodo",
    role="Ring Bearer",
    goal="Carry the One Ring to Mordor",
    tools=[aragorn_tool],
    ...
)

travel_planner = DurableAgent(
    name="TravelBuddy",
    role="Travel Planner",
    goal="Help humans find flights and remember preferences",
    instructions=[
        "Find flights to destinations",
        "Remember user preferences",
        "Provide clear flight info.",
    ],
    tools=[search_flights],
)
runner = AgentRunner()

result = await runner.run(
    travel_planner,
    payload={"task": "Plan a 3-day trip to Paris"},
)
print(result)

runner.subscribe(travel_planner)
await wait_for_shutdown()

runner.serve(
    travel_planner,
    port=8001,
)

import time

import dapr.ext.workflow as wf

wfr = wf.WorkflowRuntime()

@wfr.workflow(name="support_workflow")
def support_workflow(ctx: wf.DaprWorkflowContext, request: dict) -> str:
    triage_result = yield ctx.call_child_workflow(
        workflow="agent_workflow",
        input={"task": f"Assist with the following support request:\n\n{request}"},
        app_id="triage-agent",
    )
    if triage_result:
        print("Triage result:", triage_result.get("content", ""), flush=True)

    recommendation = yield ctx.call_child_workflow(
        workflow="agent_workflow",
        input={"task": triage_result.get("content", "")},
        app_id="expert-agent",
    )
    if recommendation:
        print("Recommendation:", recommendation.get("content", ""), flush=True)

    return recommendation.get("content", "") if recommendation else ""

wfr.start()
time.sleep(5)

client = wf.DaprWorkflowClient()
request = {
    "customer": "alice",
    "issue": "Unable to access dashboard after recent update",
}
instance_id = client.schedule_new_workflow(
    workflow=support_workflow,
    input=request,
)
client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
wfr.shutdown()

from dapr_agents import DurableAgent
from dapr_agents.agents.configs import AgentMemoryConfig
from dapr_agents.llm.dapr import DaprChatClient
from dapr_agents.memory import ConversationDaprStateMemory
from dapr_agents.workflow.runners.agent import AgentRunner

expert_agent = DurableAgent(
    name="expert_agent",
    role="Technical Support Specialist",
    goal="Provide recommendations based on customer context and issue.",
    instructions=[
        "Provide a clear, actionable recommendation to resolve the issue.",
    ],
    llm=DaprChatClient(component_name="llm-provider"),
    memory=AgentMemoryConfig(
        store=ConversationDaprStateMemory(
            store_name="agent-memory",
            session_id=f"expert-agent-session",
        )
    ),
)
runner = AgentRunner()
try:
    runner.serve(expert_agent, port=8001)
finally:
    runner.shutdown(expert_agent)

from pydantic import BaseModel
from dapr_agents.workflow.decorators.routers import message_router

class StartBlogMessage(BaseModel):
    topic: str

@message_router(
    pubsub="messagepubsub",
    topic="blog.requests",
    message_model=StartBlogMessage,
)
def blog_workflow(ctx: DaprWorkflowContext, wf_input: dict) -> str:
    outline = yield ctx.call_activity(
        create_outline, input={"topic": wf_input["topic"]}
    )
    post = yield ctx.call_activity(write_post, input={"outline": outline})
    return post

import asyncio

from dapr_agents.agents.configs import (
    AgentMemoryConfig,
    AgentProfileConfig,
    AgentPubSubConfig,
    AgentRegistryConfig,
    AgentStateConfig,
)
from dapr_agents.memory import ConversationDaprStateMemory
from dapr_agents.storage.daprstores.stateservice import StateStoreService
from dapr_agents.workflow.runners import AgentRunner
from dapr_agents.workflow.utils.core import wait_for_shutdown

registry = AgentRegistryConfig(
    store=StateStoreService(store_name="agentregistrystore"),
    team_name="fellowship",
)

frodo = DurableAgent(
    profile=AgentProfileConfig(
        name="Frodo",
        role="Ring Bearer",
        instructions=["Speak like Frodo, with humility and determination."],
    ),
    pubsub=AgentPubSubConfig(
        pubsub_name="messagepubsub",
        agent_topic="fellowship.frodo.requests",
        broadcast_topic="fellowship.broadcast",
    ),
    state=AgentStateConfig(
        store=StateStoreService(store_name="workflowstatestore", key_prefix="frodo:")
    ),
    registry=registry,
    memory=AgentMemoryConfig(
        store=ConversationDaprStateMemory(
            store_name="memorystore",
            session_id="frodo-session",
        )
    ),
)

async def main():
    runner = AgentRunner()
    try:
        runner.subscribe(frodo)
        await wait_for_shutdown()
    finally:
        runner.shutdown(frodo)

asyncio.run(main())

from dapr_agents.agents.configs import (
    AgentExecutionConfig,
    AgentPubSubConfig,
    AgentRegistryConfig,
    AgentStateConfig,
)
from dapr_agents.llm.openai import OpenAIChatClient
from dapr_agents.storage.daprstores.stateservice import StateStoreService
from dapr_agents.workflow.runners import AgentRunner
import dapr.ext.workflow as wf

llm_orchestrator = LLMOrchestrator(
    name="LLMOrchestrator",
    llm=OpenAIChatClient(),
    pubsub=AgentPubSubConfig(
        pubsub_name="messagepubsub",
        agent_topic="llm.orchestrator.requests",
        broadcast_topic="fellowship.broadcast",
    ),
    state=AgentStateConfig(
        store=StateStoreService(
            store_name="workflowstatestore", key_prefix="llm.orchestrator:"
        )
    ),
    registry=AgentRegistryConfig(
        store=StateStoreService(store_name="agentregistrystore"),
        team_name="fellowship",
    ),
    execution=AgentExecutionConfig(max_iterations=3),
    runtime=wf.WorkflowRuntime(),
)

runner = AgentRunner()
runner.serve(llm_orchestrator, port=8004)