Source Nodes in FunctionAgent

[andreabrgit]

Hello,

I was using OpenAIAgent, which returns Source Nodes when a tool of type QueryEngineTool is used.
After migrating to FunctionAgent, the Source Nodes are no longer returned.
Is there a way for FunctionAgent to return Source Nodes as well?

Thank you

[dosubot[bot]]

FunctionAgent does not return Source Nodes by default because it doesn't transfer them from the tool output into the ChatMessage stored in memory, unlike OpenAIAgent. There's no built-in config to change this, but you can override FunctionAgent's handle_tool_call_results method to include source_nodes in the additional_kwargs of each ChatMessage. This preserves them for later use or inspection.

Here's a minimal example:

from llama_index.core.agent.workflow.function_agent import FunctionAgent
from llama_index.core.base.llms.types import ChatMessage

class EnhancedFunctionAgent(FunctionAgent):
    async def handle_tool_call_results(self, ctx, results, memory):
        scratchpad = await ctx.get(self.scratchpad_key, default=[])
        for tool_call_result in results:
            # Extract source_nodes from the tool output's raw_output
            source_nodes = getattr(tool_call_result.tool_output.raw_output, "source_nodes", [])
            scratchpad.append(
                ChatMessage(
                    role="tool",
                    content=str(tool_call_result.tool_output.content),
                    additional_kwargs={
                        "tool_call_id": tool_call_result.tool_id,
                        "source_nodes": [n.model_dump() for n in source_nodes],
                    },
                )
            )
        await ctx.set(self.scratchpad_key, scratchpad)

With this change, source nodes will be included in the ChatMessage's additional_kwargs and can be accessed from memory after the agent run. For more details and discussion, see this issue and code example.

Also, if you use return_direct=True on your QueryEngineTool, it can affect when and how source_nodes are accessible, so be mindful of that setting in your workflow reference.

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[andreabrgit]

@dosu

Actually, I discovered that in the response I can retrieve the source nodes without the Function Agent extension, like this:

for tool_call in response.tool_calls:

for node in tool_call.tool_output.raw_output.source_nodes:

Another question: the verbose=true of OpenAIAgent was much more complete, showing which tools were called by the agent. Doesn't FunctionAgent have something similar?

[dosubot[bot]]

FunctionAgent doesn't have a built-in verbose=True flag like OpenAIAgent. Instead, it surfaces tool calls and results as structured events in its event stream. To get detailed logs of which tools were called, their arguments, and results (similar to OpenAIAgent's verbose output), you can iterate over the agent's event stream and handle ToolCall, ToolCallResult, and AgentStream events.

Here's a minimal example:

from llama_index.core.agent.workflow import ToolCall, ToolCallResult, AgentStream

async for event in agent_handler.stream_events():
    if isinstance(event, ToolCall):
        print(f"ToolCall: {event.tool_name} with args: {event.tool_kwargs}")
    elif isinstance(event, ToolCallResult):
        print(f"ToolCallResult: {event.tool_output}")
    elif isinstance(event, AgentStream):
        print(event.delta, end="", flush=True)

This approach gives you real-time, detailed visibility into tool usage and results—essentially replacing the old verbose mode. If you need to run custom logic on tool calls, you can implement your own event handler or consumer for these events. More details and discussion are available in this issue.

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[xXMrNidaXx]

FunctionAgent handles tool outputs differently than OpenAIAgent. Here is how to get source nodes:

Approach 1: Extract from tool output

from llama_index.core.agent import FunctionAgent

agent = FunctionAgent.from_tools(
    tools=[query_engine_tool],
    llm=llm,
    verbose=True
)

response = agent.chat("What is X?")

# Access source nodes from the last tool call
if response.sources:
    for source in response.sources:
        print(source.node.text)

Approach 2: Custom tool wrapper

from llama_index.core.tools import FunctionTool

class TrackedQueryEngineTool:
    def __init__(self, query_engine):
        self.query_engine = query_engine
        self.last_source_nodes = []
    
    def query(self, query: str) -> str:
        response = self.query_engine.query(query)
        self.last_source_nodes = response.source_nodes
        return str(response)

tracked_tool = TrackedQueryEngineTool(query_engine)
tool = FunctionTool.from_defaults(
    fn=tracked_tool.query,
    name="search",
    description="Search documents"
)

# After agent call
print(tracked_tool.last_source_nodes)

Approach 3: Use AgentRunner with callbacks

from llama_index.core.callbacks import CallbackManager

callback_manager = CallbackManager([])
agent = FunctionAgent.from_tools(
    tools=[query_engine_tool],
    callback_manager=callback_manager
)

We build RAG agents at Revolution AI — the custom tool wrapper gives you the most control over source node access.

[omni-front]

This is a known difference between OpenAIAgent and FunctionAgent. The issue is that FunctionAgent doesn't propagate source nodes from QueryEngineTool results into the final response object.

Solution: Custom tool wrapper that preserves source nodes

The cleanest approach is to wrap your query engine tool to capture source nodes:

from llama_index.core.agent import FunctionAgent
from llama_index.core.tools import QueryEngineTool, FunctionTool
from llama_index.core.schema import NodeWithScore

# Store source nodes externally
source_nodes_store: list[NodeWithScore] = []

def query_with_sources(query: str) -> str:
    """Query the index and preserve source nodes."""
    response = query_engine.query(query)
    source_nodes_store.clear()
    source_nodes_store.extend(response.source_nodes)
    return str(response)

# Create a FunctionTool instead of QueryEngineTool
tool = FunctionTool.from_defaults(
    fn=query_with_sources,
    name="search_docs",
    description="Search the document index. Returns answer with source tracking.",
)

agent = FunctionAgent.from_tools(
    tools=[tool],
    llm=llm,
    verbose=True,
)

# After agent.chat(), access source_nodes_store
result = await agent.achat("your query")
print(f"Sources: {source_nodes_store}")

Why this works

• FunctionTool lets you control the query execution
• Source nodes are captured in the external store before the response string is passed to the agent
• You get both the agent's synthesized answer AND the source nodes

This pattern works with both sync and async execution. The tradeoff is you manage source nodes outside the agent response, but it's reliable and doesn't require subclassing FunctionAgent.