执行引擎

import asyncio
from veadk import Agent, Runner

agent = Agent()
runner = Runner(agent=agent)

response = asyncio.run(runner.run("北京的天气怎么样？"))
print(response)

pip install "veadk-python[harness]==1.0.1"

from veadk import Agent, Runner
from veadk.extensions.harness.plugins import build_harness_plugins

agent = Agent(name="research_agent")
runner = Runner(
    agent=agent,
    app_name="research",
    plugins=build_harness_plugins(
        components=["invocation_context", "compactor", "response_verification"],
        profile="research",
    ),
)

export REGISTRY_TYPE="agentkit_a2a"
export REGISTRY_SPACE_ID="your-agent-center-id"
export REGISTRY_TOP_K="3"

import asyncio
from google.genai.types import Content, Part
from veadk import Agent, Runner

APP_NAME, USER_ID, SESSION_ID = "app", "user", "session"

agent = Agent()
runner = Runner(agent=agent, app_name=APP_NAME, user_id=USER_ID)

message = Content(role="user", parts=[Part(text="北京的天气怎么样？")])

async def main():
    async for event in runner.run_async(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=message,
    ):
        ...  # 见下文解析事件

asyncio.run(main())

async for event in runner.run_async(
    user_id=USER_ID, session_id=SESSION_ID, new_message=message,
):
    # 工具调用与结果
    for call in event.get_function_calls():
        print("调用工具:", call.name, call.args)
    for resp in event.get_function_responses():
        print("工具结果:", resp.name, resp.response)

    # 区分思考与文本
    if event.content:
        for part in event.content.parts:
            if part.thought:
                continue  # 模型思考过程，可按需展示
            if part.text:
                print(part.text)

    # 最终回复
    if event.is_final_response() and event.content:
        final_text = "".join(p.text for p in event.content.parts if p.text)