Add streaming support to LlmComposer.Agent with :tool_call stream chunks#117
Merged
Conversation
…port Introduces StreamCollector, the module responsible for accumulating streaming chunks of a single agent turn and reassembling them into a synthetic LlmResponse identical in shape to a non-streaming one. Supports four providers: - :open_ai / :open_router — index-based fragment merging (identical wire format; OpenRouter delegates to the same logic). - :google — already-complete FunctionCall structs, collected directly. - :bedrock — start events carry toolUseId + name; continuation events carry only inputJson fragments, grouped by toolUseId and concatenated in arrival order via a tool_id_sequence list. Also adds StreamCollector.aggregate_cost_infos/1 for summing per-turn CostInfo structs into a single run-level aggregate.
This was referenced Jun 23, 2026
sonic182
marked this pull request as ready for review
June 23, 2026 10:28
LlmComposer.Agent.run/3 now supports stream_response: true for all four providers (:open_ai, :open_router, :google, :bedrock). The loop runs intermediate tool-calling turns internally using StreamCollector and forwards only the final answer to the caller token-by-token. The stream ends with a terminal :done chunk carrying cumulative usage/cost_info and the full Agent.Result in metadata.agent_result. Hard failures yield a :error chunk instead. Adds a new :tool_call StreamChunk type: one assembled chunk per executed tool call (with result filled in) is emitted right after execution and before the next LLM turn, making the stream self-contained without needing a telemetry handler. Intermediate progress (reasoning, per-iteration data) continues to be exposed via [:llm_composer, :agent, :*] telemetry events. A new telemetry_metadata option scopes all events to a single run via an auto-generated run_id.
sonic182
force-pushed
the
feature/pr2-agent-streaming
branch
from
June 23, 2026 10:31
3839ed3 to
eba8d91
Compare
mmacia
approved these changes
Jun 23, 2026
3 tasks
JV-DM
reviewed
Jun 23, 2026
JV-DM
approved these changes
Jun 23, 2026
nvelasco
reviewed
Jun 24, 2026
nvelasco
left a comment
nvelasco
left a comment
Contributor
There was a problem hiding this comment.
@sonic182 , I don't have too much context yet.
So I was reviewing the changes with claude, here is a list of things that may worth to review it
- to_string(call.result) can crash the whole loop (Agent #111, via function_call_helpers.ex:53)
Tool results are fed back with to_string/1. A tool returning a map/list/struct (common) raises Protocol.UndefinedError and aborts
both sequential and parallel runs — contradicting the "tool failures are fed back, not aborted" promise. Confirmed untested: every
test tool returns only numbers/strings. → JSON-encode non-binary results. - :max_iterations_reached discards all accumulated work (Agent #111, agent.ex loop/4)
Returns a bare {:error, :max_iterations_reached}, throwing away cost_infos, function_calls, and the partial conversation — caller
can't see what was billed. → Return a partial Result or attach accumulated cost to the error. - Guide provider list is incomplete (#113, guides/agent.md)
Claims "OpenAI, OpenRouter, Google," but Bedrock (#98) and Ollama also support function calling and the loop is provider-agnostic. →
Say "any function-calling provider"; don't omit Bedrock/Ollama.
Worth a conscious confirmation
- Behavior change: Keyword.fetch!(:model) → Keyword.get(:model) (telemetry #112, providers_runner.ex)
Old code raised on a missing :model; new code silently yields nil in telemetry/metrics. Removes a misconfiguration guard. - functions_from_settings/1 only reads the first provider (Agent #111, agent.ex)
In a multi-provider fallback setup with per-provider functions, the default tool set can mismatch the provider actually selected at
runtime.
Minor / cosmetic
- Telemetry fires from child processes — execute_one's :telemetry.span runs inside Task.async_stream tasks (parallel mode); caller
Logger/telemetry_metadata won't auto-propagate. - :agent, :iteration, :stop has no matching :start/duration, and for tool iterations fires after tool execution — slightly
asymmetric vs. the span-based events. - O(n) list appends each iteration (cost_infos ++, messages ++, function_calls ++) — negligible at default max_iterations: 10.
- Fix to_string/1 crash in FunctionCallHelpers when tool results are maps/structs; JSON-encode non-binary results via Helpers.json_engine() with inspect/1 fallback - Return partial result (cost_infos, function_calls, messages) on :max_iterations_reached instead of discarding accumulated work; both sync and streaming paths updated - Fix functions_from_settings/1 to merge functions from all providers in multi-provider fallback setups, not just the first one - Propagate Logger.metadata into Task.async_stream workers in parallel tool execution mode - Add intentional comment on Keyword.get(:model) in providers_runner - Expand StreamCollector to support :open_ai_responses and :ollama; add merge_tool_calls/to_function_calls clauses for Responses API tool-call streaming, handling the missing call_id in delta events via current_tool_id fallback - Inject additionalProperties: false into OpenAI Responses tool schemas when strict mode is active - Log response body on non-200 errors in OpenAI Responses provider
nvelasco
reviewed
Jun 24, 2026
nvelasco
left a comment
nvelasco
left a comment
Contributor
There was a problem hiding this comment.
What about point 3 ?
- Incomplete provider list in the guide — Not addressed. The commit
updated the moduledocs in agent.ex and stream_collector.ex (now say
"all providers" + Ollama caveat), but guides/agent.md was never
touched. At the PR head it still reads:
▎ line 13–15: "The loop is synchronous (streaming is not supported
▎ in this version)* … works with any provider that supports function
▎ calling (OpenAI, OpenRouter, Google)."*
▎ line 95: "Streaming settings (stream_response: true) return
▎ {:error, :streaming_not_supported}."
This is now doubly stale: it still omits Bedrock/Ollama and claims
streaming is unsupported — directly contradicting what this entire
PR adds. Worth asking @sonic182 to update guides/agent.md to match
the new moduledocs (and the new streaming.md guide).
Member
Author
guide update (docs) is for pr3, the nex of this for usage struct, we already have a |
nvelasco
approved these changes
Jun 24, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR Series
:tool_callstream chunksSummary
Streaming agent loop:
LlmComposer.Agent.run/3now returns{:ok, stream}whensettings.stream_responseistrue. The stream yields only the final answer's:text_deltachunks followed by a terminal:donechunk with cumulativeusage/cost_infoand the fullAgent.Resultinmetadata.agent_result. Intermediate tool-calling turns run internally viaStreamCollector(PR Add LlmComposer.Agent.StreamCollector (OpenAI, OpenRouter, Google, Bedrock) #116). Supported providers::open_ai,:open_router,:open_ai_responses,:google,:bedrock,:ollama(native Ollama does not emit tool-call deltas; use:open_aipointed at Ollama's OpenAI-compatible endpoint for tool-call streaming).:tool_callstream chunk type: After executing tool calls the agent emits one assembled%StreamChunk{type: :tool_call}chunk per call (result already filled in) before starting the next LLM turn. This makes the stream self-contained — callers can react to tool activity inline without attaching a telemetry handler:Telemetry:
[:llm_composer, :agent, :tool, :start]now includes:arguments,:metadata, and:id. A new[:llm_composer, :agent, :reasoning, :delta]event surfaces intermediate reasoning. Passtelemetry_metadata:torun/3to scope all events to a single run via an auto-generated:run_id.Dialyzer fix: removed unreachable fallback clause in
user_prompt/2(Settings.user_prompt_prefixis always aString.t()).Bug fixes
Tool result encoding crash:
FunctionCallHelpers.build_tool_result_messages/1usedto_string/1on tool results, raisingProtocol.UndefinedErrorfor any tool returning a map, list, or struct. Results are now JSON-encoded viaHelpers.json_engine()with aninspect/1fallback.:max_iterations_reacheddiscards accumulated work: the loop returned a bare{:error, :max_iterations_reached}, throwing awaycost_infos,function_calls, and the partial conversation. Now returns{:error, {:max_iterations_reached, %{cost_infos, function_calls, messages, iterations}}}. The streaming path includes the same data in the error chunk metadata.functions_from_settings/1only read the first provider: in a multi-provider fallback setup, only the first provider's:functionswere used. Now merges functions from all providers and deduplicates by name.Logger metadata not propagated in parallel tool execution:
Task.async_streamworkers in:parallelmode didn't inherit the caller'sLogger.metadata/0. Parent metadata is now captured and restored inside each worker.Test plan
mix test test/llm_composer/agent_test.exsmix precommitsamples/loop_agent_all_providers_stream.ex