使用结构化输出

大多数时候，你需要智能体以特定格式返回结果。智能体可以通过两种方式返回结构化JSON：

output_clsoutput_cls – 用作输出模式的Pydantic模型
structured_output_fnstructured_output_fn – 针对更高级的用例，提供一个自定义函数，用于验证或重写智能体的对话以适应您所需的任何模型。

无论是像 FunctionAgent 和 ReActAgent 这样的单智能体，还是多智能体 AgentWorkflow 工作流，都支持这些选项——让我们探索各种可能性：

使用 structured_output_fn`output_cls`

from llama_index.core.agent.workflow import FunctionAgent, AgentWorkflow
from llama_index.llms.openai import OpenAI
from pydantic import BaseModel, Field

llm = OpenAI(model="gpt-4.1")


## define structured output format  and tools
class MathResult(BaseModel):
    operation: str = Field(description="the performed operation")
    result: int = Field(description="the result of the operation")


def multiply(x: int, y: int):
    """Multiply two numbers"""
    return x * y


## define agent
agent = FunctionAgent(
    tools=[multiply],
    name="calculator",
    system_prompt="You are a calculator agent who can multiply two numbers using the `multiply` tool.",
    output_cls=MathResult,
    llm=llm,
)

response = await agent.run("What is 3415 * 43144?")
print(response.structured_response)
print(response.get_pydantic_model(MathResult))

这也适用于多智能体工作流：

## define structured output format  and tools
class Weather(BaseModel):
    location: str = Field(description="The location")
    weather: str = Field(description="The weather")


def get_weather(location: str):
    """Get the weather for a given location"""
    return f"The weather in {location} is sunny"


## define single agents
agent = FunctionAgent(
    llm=llm,
    tools=[get_weather],
    system_prompt="You are a weather agent that can get the weather for a given location",
    name="WeatherAgent",
    description="The weather forecaster agent.",
)
main_agent = FunctionAgent(
    name="MainAgent",
    tools=[],
    description="The main agent",
    system_prompt="You are the main agent, your task is to dispatch tasks to secondary agents, specifically to WeatherAgent",
    can_handoff_to=["WeatherAgent"],
    llm=llm,
)

## define multi-agent workflow
workflow = AgentWorkflow(
    agents=[main_agent, agent],
    root_agent=main_agent.name,
    output_cls=Weather,
)

response = await workflow.run("What is the weather in Tokyo?")
print(response.structured_response)
print(response.get_pydantic_model(Weather))

使用 structured_output_fn`structured_output_fn`

自定义函数应接收由智能体工作流生成的一系列ChatMessage对象作为输入，并返回一个字典（可转换为BaseModel子类）：

import json
from llama_index.core.llms import ChatMessage
from typing import List, Dict, Any


class Flavor(BaseModel):
    flavor: str
    with_sugar: bool


async def structured_output_parsing(
    messages: List[ChatMessage],
) -> Dict[str, Any]:
    sllm = llm.as_structured_llm(Flavor)
    messages.append(
        ChatMessage(
            role="user",
            content="Given the previous message history, structure the output based on the provided format.",
        )
    )
    response = await sllm.achat(messages)
    return json.loads(response.message.content)


def get_flavor(ice_cream_shop: str):
    return "Strawberry with no extra sugar"


agent = FunctionAgent(
    tools=[get_flavor],
    name="ice_cream_shopper",
    system_prompt="You are an agent that knows the ice cream flavors in various shops.",
    structured_output_fn=structured_output_parsing,
    llm=llm,
)

response = await agent.run(
    "What strawberry flavor is available at Gelato Italia?"
)
print(response.structured_response)
print(response.get_pydantic_model(Flavor))

流式传输结构化输出

您可以在工作流运行时通过使用 AgentStreamStructuredOutput 事件获取结构化输出：

from llama_index.core.agent.workflow import (
    AgentInput,
    AgentOutput,
    ToolCall,
    ToolCallResult,
    AgentStreamStructuredOutput,
)

handler = agent.run("What strawberry flavor is available at Gelato Italia?")

async for event in handler.stream_events():
    if isinstance(event, AgentInput):
        print(event)
    elif isinstance(event, AgentStreamStructuredOutput):
        print(event.output)
        print(event.get_pydantic_model(Weather))
    elif isinstance(event, ToolCallResult):
        print(event)
    elif isinstance(event, ToolCall):
        print(event)
    elif isinstance(event, AgentOutput):
        print(event)
    else:
        pass

response = await handler

你可以通过直接作为字典访问或使用 get_pydantic_model 方法将其加载为 BaseModel 子类来解析智能体响应中的结构化输出：

print(response.structured_response)
print(response.get_pydantic_model(Flavor))

使用结构化输出

使用 structured_output_fnoutput_cls

使用 structured_output_fnstructured_output_fn

流式传输结构化输出

使用 structured_output_fn`output_cls`

使用 structured_output_fn`structured_output_fn`