Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions apps/website/content/docs/langgraph/api/api-docs.json
Original file line number Diff line number Diff line change
Expand Up @@ -946,6 +946,12 @@
"description": "Custom message deserializer for non-standard message formats.",
"optional": true
},
{
"name": "transcriptNodeNames",
"type": "string[]",
"description": "LangGraph node names whose `messages-tuple` LLM chunks should be projected\ninto the main chat transcript. Omit to accept all top-level message chunks.",
"optional": true
},
{
"name": "transport",
"type": "AgentTransport",
Expand Down Expand Up @@ -1082,6 +1088,12 @@
"description": "Custom message deserializer for non-standard message formats.",
"optional": true
},
{
"name": "transcriptNodeNames",
"type": "string[]",
"description": "LangGraph node names whose `messages-tuple` LLM chunks should be projected\ninto the main chat transcript. Omit to accept all top-level message chunks.\n\nUse this when a graph has side-effect LLM nodes, such as title generation,\nwhose streamed model output should not render as assistant chat content.",
"optional": true
},
{
"name": "transport",
"type": "AgentTransport",
Expand Down
13 changes: 13 additions & 0 deletions apps/website/content/docs/langgraph/api/provide-agent.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ bootstrapApplication(AppComponent, {
| `telemetry` | `AgentRuntimeTelemetrySink \| false` | Optional app-owned telemetry sink. No telemetry is emitted unless this is provided. |
| `filterSubagentMessages` | `boolean` | When true, subagent messages are filtered from the main messages signal. |
| `subagentToolNames` | `string[]` | Tool names that indicate a subagent invocation. |
| `transcriptNodeNames` | `string[]` | LangGraph node names whose `messages-tuple` chunks should stream into the main chat transcript. Omit to accept all top-level chunks. |

## Singleton model

Expand All @@ -56,6 +57,18 @@ provideAgent({
const chat = injectAgent();
```

## Transcript node filtering

LangGraph streams `messages-tuple` chunks for every LLM node in a run. If your graph has side-effect LLM nodes, such as a title generator or evaluator, set `transcriptNodeNames` so only your conversational node updates `messages()`.

```ts
provideAgent({
apiUrl: 'https://api.example.com',
assistantId: 'support-agent',
transcriptNodeNames: ['generate'],
});
```

## Test transports

`transport` is an object that implements `AgentTransport`, not an Angular class token. Create an instance before passing it to `provideAgent()`.
Expand Down
12 changes: 3 additions & 9 deletions examples/chat/angular/e2e/aimock-runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,9 @@ function loadFixtureEntries(fixturePath: string): FixtureFileEntry[] {
export async function startAimock(opts: AimockStartOptions): Promise<AimockHandle> {
const entries = loadFixtureEntries(opts.fixturePath);

// Use a large chunkSize so each response arrives in 1-2 SSE deltas. This
// intentionally turns off the partial-markdown streaming path for harness
// tests: structural assertions (code fence, list) measure the FINAL rendered
// DOM, not the progressive render. With aggressive default chunking, the
// partial-markdown parser sometimes can't recover a triple-backtick fence
// that gets split mid-token, and the final state ends up as inline <code>
// instead of <pre><code>. Streaming-progressive behavior is covered by the
// Phase 1 unit-variance tables; the e2e harness is for final-state
// invariants and cross-stack integration.
// Use a large default chunkSize so ordinary fixture responses arrive in 1-2
// SSE deltas. Most e2e assertions measure the final rendered DOM, while
// targeted streaming regressions opt into smaller per-fixture chunks.
const mock = new LLMock({ port: 0, chunkSize: 4096 });
if (entries.length > 0) {
mock.addFixturesFromJSON(entries as never);
Expand Down
28 changes: 28 additions & 0 deletions examples/chat/angular/e2e/fixtures/streaming-markdown.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
{
"fixtures": [
{
"match": { "userMessage": "stream a markdown comparison table regression" },
"response": {
"content": "Here is the comparison:\n\n| Name | Mental model | When to use |\n| --- | --- | --- |\n| Angular Signals | Synchronous value graph | Local component state |\n| RxJS | Event stream | Async flows and cancellation |\n| zone.js | Async task patching | Zone-based change detection |\n\nDone."
},
"chunkSize": 4,
"latency": 75
},
{
"match": { "userMessage": "stream a blockquote then a markdown table regression" },
"response": {
"content": "> First line of the quote.\n> Second line of the quote.\n\n| Issue | Expected behavior | Verification |\n| --- | --- | --- |\n| Button click does not update UI | UI reflects new state immediately | Click button and observe state |\n| Slow initial render | Main content appears within target | Measure first meaningful paint |"
},
"chunkSize": 6,
"latency": 25
},
{
"match": { "userMessage": "stream a TypeScript code fence regression" },
"response": {
"content": "Here is the snippet:\n\n```typescript\nconst answer = 42;\n```\n\nThe constant is available for later use."
},
"chunkSize": 3,
"latency": 35
}
]
}
150 changes: 148 additions & 2 deletions examples/chat/angular/e2e/markdown-surfaces.spec.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,13 @@
// SPDX-License-Identifier: MIT
import { test, expect, type Locator } from '@playwright/test';
import { sendPromptAndWait } from './test-helpers';
import { test, expect, type Locator, type Page } from '@playwright/test';
import {
attachBrowserHygiene,
messageInput,
openDemo,
sendButton,
sendPromptAndWait,
waitForFinalAssistant,
} from './test-helpers';

test('heading: assistant bubble renders an <h1>', async ({ page }) => {
const bubble = await sendPromptAndWait(page, 'respond with a heading');
Expand Down Expand Up @@ -55,6 +62,70 @@ test('markdown checklist matrix: rich markdown renders with escaped html', async
await expect(bubble).toContainText("<script>alert('xss')</script>");
});

test('streaming markdown table: keeps in-progress rows inside one table', async ({ page }) => {
const hygiene = attachBrowserHygiene(page);
await sendPrompt(page, 'stream a markdown comparison table regression');

const streamingAssistant = latestAssistant(page);
await expect(streamingAssistant).toBeAttached({ timeout: 15_000 });

const samples = await collectStreamingSamples(page, streamingAssistant, 2_000);
expect(contentChangedAcross(samples)).toBe(true);
const tableSamples = samples.filter((sample) => sample.tableCount > 0);
expect(tableSamples.length).toBeGreaterThan(2);
expect(tableSamples.every((sample) => sample.tableCount <= 1)).toBe(true);
expect(tableSamples.every((sample) => sample.rowsOutsideTable === 0)).toBe(true);
expect(tableSamples.every((sample) => sample.detachedTableCellText.length === 0)).toBe(true);

const bubble = await waitForFinalAssistant(page);
await expect(bubble.locator('table')).toHaveCount(1);
await expect(bubble.locator('thead th')).toHaveText(['Name', 'Mental model', 'When to use']);
await expect(bubble.locator('tbody tr')).toHaveCount(3);
await expect(bubble.locator('tbody tr').nth(2)).toContainText('zone.js');
await expect.poll(async () => tableColumnsAlign(bubble)).toBe(true);
expect(hygiene.consoleErrors).toEqual([]);
expect(hygiene.failedRequests).toEqual([]);
});

test('streaming markdown table: blockquote followed by table does not throw', async ({ page }) => {
const hygiene = attachBrowserHygiene(page);
await sendPrompt(page, 'stream a blockquote then a markdown table regression');

const bubble = await waitForFinalAssistant(page);
await expect(bubble.locator('blockquote')).toBeVisible();
await expect(bubble.locator('blockquote')).toContainText('First line of the quote.');
await expect(bubble.locator('table')).toHaveCount(1);
await expect(bubble.locator('thead th')).toHaveText([
'Issue',
'Expected behavior',
'Verification',
]);
await expect(bubble.locator('tbody tr')).toHaveCount(2);
expect(hygiene.consoleErrors).toEqual([]);
expect(hygiene.failedRequests).toEqual([]);
});

test('streaming code fence: suppresses closing fence marker while streaming', async ({ page }) => {
const hygiene = attachBrowserHygiene(page);
await sendPrompt(page, 'stream a TypeScript code fence regression');

const streamingAssistant = latestAssistant(page);
await expect(streamingAssistant).toBeAttached({ timeout: 15_000 });

const samples = await collectStreamingSamples(page, streamingAssistant, 4_000);
expect(contentChangedAcross(samples)).toBe(true);
const codeSamples = samples.filter((sample) => sample.codeBlockCount > 0);
expect(codeSamples.length).toBeGreaterThan(2);
expect(codeSamples.every((sample) => !sample.hasRawFenceMarker)).toBe(true);

const bubble = await waitForFinalAssistant(page);
await expect(bubble.locator('pre code')).toHaveCount(1);
await expect(bubble.locator('pre code')).toContainText('const answer = 42');
expect(await bubbleContainsRawFenceMarker(bubble)).toBe(false);
expect(hygiene.consoleErrors).toEqual([]);
expect(hygiene.failedRequests).toEqual([]);
});

async function tableColumnsAlign(bubble: Locator): Promise<boolean> {
const table = bubble.locator('table').first();
return table.evaluate((el) => {
Expand All @@ -72,3 +143,78 @@ async function tableColumnsAlign(bubble: Locator): Promise<boolean> {
});
});
}

async function sendPrompt(page: Page, prompt: string): Promise<void> {
await openDemo(page, '/embed');
await messageInput(page).fill(prompt);
await sendButton(page).click();
}

function latestAssistant(page: Page): Locator {
return page.locator('chat-message[data-role="assistant"]').last();
}

interface StreamingMarkdownSample {
readonly contentTextLength: number;
readonly tableCount: number;
readonly rowsOutsideTable: number;
readonly detachedTableCellText: string[];
readonly codeBlockCount: number;
readonly hasRawFenceMarker: boolean;
}

async function collectStreamingSamples(
page: Page,
bubble: Locator,
minimumDurationMs: number,
): Promise<StreamingMarkdownSample[]> {
const samples: StreamingMarkdownSample[] = [];
const startedAt = Date.now();

do {
samples.push(await sampleStreamingMarkdown(bubble));
await page.waitForTimeout(75);
} while (Date.now() - startedAt < minimumDurationMs);

return samples;
}

async function sampleStreamingMarkdown(bubble: Locator): Promise<StreamingMarkdownSample> {
return bubble.evaluate((el) => {
const looksLikeTableRowFragment = (text: string): boolean => (
/\|/.test(text) || /Angular Signals|RxJS|zone\.js/.test(text)
);
const tables = Array.from(el.querySelectorAll('table'));
const rowsOutsideTable = Array.from(el.querySelectorAll('tr')).filter(
(row) => !row.closest('table'),
).length;
const detachedTableCellText: string[] = [];
const walker = document.createTreeWalker(el, NodeFilter.SHOW_TEXT);
let node = walker.nextNode();
while (node) {
const parent = node.parentElement;
const text = node.textContent?.trim() ?? '';
if (text && !parent?.closest('table') && looksLikeTableRowFragment(text)) {
detachedTableCellText.push(text);
}
node = walker.nextNode();
}

return {
contentTextLength: el.textContent?.length ?? 0,
tableCount: tables.length,
rowsOutsideTable,
detachedTableCellText,
codeBlockCount: el.querySelectorAll('pre code').length,
hasRawFenceMarker: el.textContent?.includes('```') ?? false,
};
});
}

async function bubbleContainsRawFenceMarker(bubble: Locator): Promise<boolean> {
return bubble.evaluate((el) => el.textContent?.includes('```') ?? false);
}

function contentChangedAcross(samples: StreamingMarkdownSample[]): boolean {
return new Set(samples.map((sample) => sample.contentTextLength)).size > 2;
}
1 change: 1 addition & 0 deletions examples/chat/angular/e2e/test-helpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ export function attachBrowserHygiene(page: Page): {
if (msg.type() !== 'error') return;
const text = msg.text();
if (/PostHog|ERR_NAME_NOT_RESOLVED|license/i.test(text)) return;
if (/409 \(Conflict\)/i.test(text)) return;
consoleErrors.push(text);
});
page.on('pageerror', (err) => {
Expand Down
3 changes: 3 additions & 0 deletions examples/chat/angular/src/app/shell/demo-shell.component.ts
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,9 @@ export function isConflict(err: unknown): boolean {
// subagent dispatches and to materialize agent.subagents() from the
// resulting tools:<id>-namespaced stream events.
subagentToolNames: ['research'],
// The canonical graph has side-effect LLM nodes such as generate_title;
// only the generate node's token stream belongs in the chat transcript.
transcriptNodeNames: ['generate'],
telemetry: (event) => telemetrySink?.(event),
}),
{ provide: DEMO_AGENT, useFactory: () => inject(DemoShell).agent },
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,30 @@ describe('ChatStreamingMdComponent — streaming table rendering', () => {
}
});

it('streams a realistic comparison table as one table across small chunks', () => {
host.streaming.set(true);
const content =
'Here is the comparison:\n\n' +
'| Name | Mental model | When to use |\n' +
'| --- | --- | --- |\n' +
'| Angular Signals | Synchronous value graph | Local component state |\n' +
'| RxJS | Event stream | Async flows and cancellation |\n' +
'| zone.js | Async task patching | Zone-based change detection |\n\n' +
'Done.';

for (let i = 6; i <= content.length; i += 6) {
grow(content.slice(0, i));
const tableCount = el.querySelectorAll('table').length;
if (tableCount > 0) {
expect(tableCount, `one table at ${JSON.stringify(content.slice(0, i).slice(-40))}`).toBe(1);
}
expect(
[...el.querySelectorAll('p')].some((p) => (p.textContent || '').includes('| zone.js')),
`no detached row paragraph at ${JSON.stringify(content.slice(0, i).slice(-40))}`,
).toBe(false);
}
});

it('keeps a finalized partial body row in the table when the stream pauses', () => {
vi.useFakeTimers();
try {
Expand Down
6 changes: 6 additions & 0 deletions libs/langgraph/src/lib/agent.provider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,11 @@ export interface AgentConfig<
filterSubagentMessages?: boolean;
/** Tool names that indicate a subagent invocation. */
subagentToolNames?: string[];
/**
* LangGraph node names whose `messages-tuple` LLM chunks should be projected
* into the main chat transcript. Omit to accept all top-level message chunks.
*/
transcriptNodeNames?: string[];
}

/**
Expand Down Expand Up @@ -84,6 +89,7 @@ function agentFactory<T>(): LangGraphAgent<T> {
...(config.telemetry !== undefined ? { telemetry: config.telemetry } : {}),
...(config.filterSubagentMessages !== undefined ? { filterSubagentMessages: config.filterSubagentMessages } : {}),
...(config.subagentToolNames !== undefined ? { subagentToolNames: config.subagentToolNames } : {}),
...(config.transcriptNodeNames !== undefined ? { transcriptNodeNames: config.transcriptNodeNames } : {}),
});
}

Expand Down
8 changes: 8 additions & 0 deletions libs/langgraph/src/lib/agent.types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,14 @@ export interface AgentOptions<T, _ResolvedBag extends BagTemplate> {
filterSubagentMessages?: boolean;
/** Tool names that indicate a subagent invocation. */
subagentToolNames?: string[];
/**
* LangGraph node names whose `messages-tuple` LLM chunks should be projected
* into the main chat transcript. Omit to accept all top-level message chunks.
*
* Use this when a graph has side-effect LLM nodes, such as title generation,
* whose streamed model output should not render as assistant chat content.
*/
transcriptNodeNames?: string[];
}

// ── SubagentStreamRef ────────────────────────────────────────────────────────
Expand Down
26 changes: 24 additions & 2 deletions libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1459,12 +1459,12 @@ describe('stream-manager.bridge — captured streaming replay (Finding C)', () =
describe('identity-based delta merge (messages-tuple)', () => {
const META = { langgraph_node: 'chatbot' };

function setup() {
function setup(extraOptions: Record<string, unknown> = {}) {
const transport = new MockAgentTransport();
const subjects = makeSubjects();
const destroy$ = new Subject<void>();
const bridge = createStreamManagerBridge({
options: { apiUrl: '', assistantId: 'test', transport },
options: { apiUrl: '', assistantId: 'test', transport, ...extraOptions },
subjects,
threadId$: of(null),
destroy$: destroy$.asObservable(),
Expand Down Expand Up @@ -1586,6 +1586,28 @@ describe('identity-based delta merge (messages-tuple)', () => {
destroy$.next();
});

it('ignores message tuple chunks from non-transcript nodes when configured', async () => {
const { transport, subjects, destroy$, bridge } = setup({
transcriptNodeNames: ['chatbot'],
});
bridge.submit({});
transport.emit([
tupleEvent('ai-1', 'Here is the answer.'),
{
type: 'messages',
data: [
{ id: 'title-1', type: 'ai', content: 'This title must not render.' },
{ langgraph_node: 'generate_title' },
],
messageMetadata: { langgraph_node: 'generate_title' },
} as any,
]);
transport.close();
await new Promise(r => setTimeout(r, 10));
expect(lastAiContent(subjects)).toBe('Here is the answer.');
destroy$.next();
});

it('values-sync mid-run keeps snapshot semantics (lagging state does not rewind)', async () => {
const { transport, subjects, destroy$, bridge } = setup();
bridge.submit({});
Expand Down
Loading
Loading