执行引擎

import asyncio
from veadk import Agent, Runner

agent = Agent(name="assistant")
runner = Runner(agent=agent, app_name="demo", user_id="demo-user")

async def main():
    response = await runner.run(
        "用一句话解释智能体的用途", session_id="demo-session"
    )
    print(response)

asyncio.run(main())

import asyncio
from google.adk.agents.run_config import RunConfig, StreamingMode
from google.genai.types import Content, Part
from veadk import Agent, Runner

APP_NAME, USER_ID, SESSION_ID = "stream_demo", "demo-user", "demo-session"
runner = Runner(
    agent=Agent(name="assistant"), app_name=APP_NAME, user_id=USER_ID
)

async def main():
    await runner.session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
    message = Content(role="user", parts=[Part(text="用一句话解释智能体的用途")])
    async for event in runner.run_async(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=message,
        run_config=RunConfig(streaming_mode=StreamingMode.SSE),
    ):
        if event.error_code:
            print("Error:", event.error_code, event.error_message)
        for call in event.get_function_calls():
            print("Tool call:", call.name)
        for result in event.get_function_responses():
            print("Tool result:", result.name)
        if event.partial:
            continue
        if event.is_final_response() and event.content:
            text = "".join(
                part.text for part in (event.content.parts or [])
                if part.text and not part.thought
            )
            if text:
                print(text)

asyncio.run(main())