Implementation:Run llama Llama index AgentOutput Processing

Overview

AgentOutput Processing documents the data structures and patterns used to access and interpret ReAct agent results in LlamaIndex. The primary output type is AgentOutput, which contains the response message, tool call records, and optional structured output. Supporting types include ToolCallResult, ToolOutput, AgentStream, and ToolCall.

Principle:Run_llama_Llama_index_Agent_Output_Processing

Source Files

• llama-index-core/llama_index/core/agent/workflow/workflow_events.py -- Event and output type definitions
• llama-index-core/llama_index/core/tools/types.py -- ToolOutput definition
• llama-index-core/llama_index/core/agent/workflow/react_agent.py -- Finalization logic

AgentOutput

Source: workflow_events.py, Lines 70-93

class AgentOutput(Event):
    """LLM output."""

    response: ChatMessage
    structured_response: Optional[Dict[str, Any]] = Field(default=None)
    current_agent_name: str
    raw: Optional[Any] = Field(default=None, exclude=True)
    tool_calls: list[ToolSelection] = Field(default_factory=list)
    retry_messages: list[ChatMessage] = Field(default_factory=list)

Fields

Field	Type	Description
response	ChatMessage	The final response message from the agent. Access the text via response.content or individual blocks via response.blocks.
structured_response	Optional[Dict[str, Any]]	Structured output dictionary, populated when output_cls or structured_output_fn is configured.
current_agent_name	str	The name of the agent that produced this output.
raw	Optional[Any]	The raw response object from the LLM provider (excluded from serialization).
tool_calls	list[ToolSelection]	All tool selections made during the agent's run. After finalization, this includes the accumulated ToolCallResult objects.
retry_messages	list[ChatMessage]	Internal field used for error recovery. Non-empty when the LLM output could not be parsed and needs to be retried.

Key Methods

# String representation returns the response content
def __str__(self) -> str:
    return self.response.content or ""

# Convert structured_response to a Pydantic model
def get_pydantic_model(self, model: Type[BaseModel]) -> Optional[BaseModel]:
    """Validates and converts structured_response to the given Pydantic model.
    Returns None with a warning if validation fails."""

ToolCallResult

Source: workflow_events.py, Lines 104-111

class ToolCallResult(Event):
    """Tool call result."""

    tool_name: str
    tool_kwargs: dict
    tool_id: str
    tool_output: ToolOutput
    return_direct: bool

Field	Type	Description
tool_name	str	The name of the tool that was called.
tool_kwargs	dict	The keyword arguments passed to the tool.
tool_id	str	UUID identifying this specific tool invocation.
tool_output	ToolOutput	The tool's output, containing content, raw data, and error status.
return_direct	bool	Whether the tool output should be returned directly to the user without further LLM processing.

ToolOutput

Source: tools/types.py, Lines 106-165

class ToolOutput(BaseModel):
    """Tool output."""

    blocks: List[ContentBlock]
    tool_name: str
    raw_input: Dict[str, Any]
    raw_output: Any
    is_error: bool = False

Field / Property	Type	Description
blocks	List[ContentBlock]	Content blocks (TextBlock, ImageBlock, AudioBlock, etc.) containing the output.
content (property)	str	Concatenated text from all TextBlock entries. Settable -- replaces blocks with a single TextBlock.
tool_name	str	Name of the tool that produced this output.
raw_input	Dict[str, Any]	The original arguments passed to the tool.
raw_output	Any	The unprocessed return value of the underlying function.
is_error	bool	True if the tool call raised an exception.
exception (property)	Optional[Exception]	The exception object if is_error is True (stored as a private attribute).

ToolCall

Source: workflow_events.py, Lines 96-101

class ToolCall(Event):
    """All tool calls are surfaced."""

    tool_name: str
    tool_kwargs: dict
    tool_id: str

This event is emitted before a tool is executed, allowing streaming consumers to show "calling tool X..." messages. The corresponding ToolCallResult is emitted after execution.

AgentStream

Source: workflow_events.py, Lines 38-46

class AgentStream(Event):
    """Agent stream."""

    delta: str
    response: str
    current_agent_name: str
    tool_calls: list[ToolSelection] = Field(default_factory=list)
    raw: Optional[Any] = Field(default=None, exclude=True)
    thinking_delta: Optional[str] = Field(default=None)

Field	Type	Description
delta	str	The incremental token(s) just generated by the LLM.
response	str	The accumulated response text so far.
current_agent_name	str	Which agent is currently generating.
tool_calls	list[ToolSelection]	Tool calls parsed from the stream so far (for function-calling agents).
raw	Optional[Any]	Raw LLM response chunk (excluded from serialization).
thinking_delta	Optional[str]	Thinking/reasoning tokens for models that support extended thinking.

ReActAgent Finalization

Source: react_agent.py, Lines 289-326

The finalize method performs post-processing on the agent output before it is returned:

async def finalize(self, ctx, output, memory) -> AgentOutput:
    current_reasoning = await ctx.store.get(self.reasoning_key, default=[])

    if len(current_reasoning) > 0 and isinstance(current_reasoning[-1], ResponseReasoningStep):
        # Serialize full reasoning chain to memory
        reasoning_str = "\n".join([x.get_content() for x in current_reasoning])
        if reasoning_str:
            reasoning_msg = ChatMessage(role="assistant", content=reasoning_str)
            await memory.aput(reasoning_msg)

        # Strip "Answer:" prefix from response text
        text_block = None
        for block in output.response.blocks:
            if isinstance(block, TextBlock):
                text_block = block
                break
        if text_block and "Answer:" in text_block.text:
            start_idx = text_block.text.find("Answer:")
            text_block.text = text_block.text[start_idx + len("Answer:"):].strip()

        # Clear reasoning chain
        await ctx.store.set(self.reasoning_key, [])

    return output

Usage Examples

Basic Output Access

from llama_index.core.agent.workflow import ReActAgent

agent = ReActAgent(tools=[multiply_tool, search_tool], llm=llm)
handler = agent.run("What is 6 times 7?")
result = await handler

# Access the response text
print(str(result))               # "42"
print(result.response.content)   # "42"

# Access tool calls made during execution
for tool_call in result.tool_calls:
    print(f"Tool: {tool_call.tool_name}, Args: {tool_call.tool_kwargs}")

Streaming with Event Processing

from llama_index.core.agent.workflow.workflow_events import (
    AgentOutput, AgentStream, ToolCall, ToolCallResult,
)

handler = agent.run("Research and summarize AI trends.")

async for event in handler.stream_events():
    if isinstance(event, AgentStream):
        print(event.delta, end="", flush=True)
    elif isinstance(event, ToolCall):
        print(f"\n[Calling: {event.tool_name}({event.tool_kwargs})]")
    elif isinstance(event, ToolCallResult):
        if event.tool_output.is_error:
            print(f"\n[Error in {event.tool_name}: {event.tool_output.content}]")
        else:
            print(f"\n[{event.tool_name} returned: {event.tool_output.content[:80]}...]")

result = await handler
print(f"\nFinal answer: {result}")

Structured Output

from pydantic import BaseModel

class ResearchResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float

agent = ReActAgent(tools=[search_tool], llm=llm, output_cls=ResearchResult)
handler = agent.run("What are the latest trends in renewable energy?")
result = await handler

# Access structured output
structured = result.get_pydantic_model(ResearchResult)
if structured is not None:
    print(f"Summary: {structured.summary}")
    print(f"Confidence: {structured.confidence}")
    for finding in structured.key_findings:
        print(f"  - {finding}")
else:
    # Fallback to text response
    print(result.response.content)

Inspecting Tool Call History

handler = agent.run("Compare the GDP of USA and China.")
result = await handler

# Iterate over all tool calls
for tc in result.tool_calls:
    if hasattr(tc, 'tool_output'):
        # This is a ToolCallResult (accumulated during execution)
        print(f"Called {tc.tool_name} with {tc.tool_kwargs}")
        print(f"  Output: {tc.tool_output.content[:200]}")
        print(f"  Error: {tc.tool_output.is_error}")
    else:
        # This is a ToolSelection (tool_name, tool_kwargs, tool_id)
        print(f"Selected {tc.tool_name} with {tc.tool_kwargs}")

Import Statements

from llama_index.core.agent.workflow import AgentOutput
from llama_index.core.agent.workflow.workflow_events import (
    AgentOutput,
    AgentStream,
    ToolCall,
    ToolCallResult,
)
from llama_index.core.tools.types import ToolOutput