diff --git a/CHANGELOG.md b/CHANGELOG.md index 94e0f6d0..0ffb6b76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ - New `human_decide` graph node emits AG-UI tool calls (`TOOL_CALL_START/ARGS/END`) for frontend integration - New `emit_tool_call_start()`, `emit_tool_call_args()`, `emit_tool_call_end()` AG-UI event helpers - New `AGUIEmitter.emit()` method for direct event emission +- **AG-UI Research Example**: Updated with interactive decision UI + - Decision panel with question editing (add/remove) at each decision point - **HotpotQA Evaluation**: Added HotpotQA dataset adapter for multi-hop QA benchmarks - Extracts unique documents from validation set context paragraphs - Uses MAP for retrieval evaluation (multiple supporting documents per question) diff --git a/docs/agents.md b/docs/agents.md index 103afa9c..69bc0e82 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -219,4 +219,8 @@ In interactive mode, you can: - Execute searches and review collected answers - Continue researching or synthesize when ready -For a web-based interactive experience, see the [AG-UI Research Example](https://github.com/ggozad/haiku.rag/tree/main/examples/ag-ui-research). +For a web-based interactive experience with visual decision UI, see the [AG-UI Research Example](https://github.com/ggozad/haiku.rag/tree/main/examples/ag-ui-research). The web interface provides: + +- Question editing panel to add/remove sub-questions at decision points +- Search and Generate Report buttons for controlling research flow +- Live state display showing answers, confidence, and progress diff --git a/examples/ag-ui-research/README.md b/examples/ag-ui-research/README.md index df6850dc..1fa202c9 100644 --- a/examples/ag-ui-research/README.md +++ b/examples/ag-ui-research/README.md @@ -1,13 +1,13 @@ # Interactive Research Assistant -Research assistant powered by [haiku.rag](https://ggozad.github.io/haiku.rag/), [Pydantic Graph](https://ai.pydantic.dev/graph/), and [AG-UI](https://docs.ag-ui.com/). Ask complex questions and watch the research process unfold in real-time. +Research assistant powered by [haiku.rag](https://ggozad.github.io/haiku.rag/), [Pydantic Graph](https://ai.pydantic.dev/graph/), and [AG-UI](https://docs.ag-ui.com/). Ask complex questions and watch the research process unfold in real-time with human-in-the-loop control. [Watch demo video](https://vimeo.com/1128874386) ## Features -- **Multi-iteration research graph**: Automated question decomposition and search -- **Intelligent evaluation**: Confidence-based decision making with automatic iteration until sufficient information is gathered +- **Human-in-the-loop research**: Review and modify questions at decision points, then continue searching or generate report +- **Multi-iteration research graph**: Automated question decomposition and parallel search - **Live state synchronization**: Real-time delta updates of research progress via AG-UI protocol - **Rich reporting**: Generates comprehensive research reports with findings, conclusions, and sources @@ -25,9 +25,7 @@ Research assistant powered by [haiku.rag](https://ggozad.github.io/haiku.rag/), **Option A: Create a new database** ```bash - mkdir -p data - haiku-rag add "Your documents here" --db data/haiku_rag.lancedb - # Or add from files + haiku-rag init --db data/haiku_rag.lancedb haiku-rag add-src document.pdf --db data/haiku_rag.lancedb ``` @@ -63,27 +61,29 @@ Research assistant powered by [haiku.rag](https://ggozad.github.io/haiku.rag/), DB_PATH=/path/to/your/existing/haiku_rag.lancedb # If using an existing db. ``` -1. **Start the application** +4. **Start the application** ```bash docker compose up --build ``` -2. **Access the interface** +5. **Access the interface** - Frontend: http://localhost:3000 - Backend health: http://localhost:8000/health ## How It Works 1. **Ask a question**: Type your research question in the chat -2. **Plan phase**: The research graph automatically: - - Decomposes your question into targeted sub-questions - - Gathers initial context about the topic -3. **Research iterations**: The graph autonomously: - - Searches the knowledge base for each sub-question in parallel - - Assesses confidence in gathered information - - Generates new follow-up questions if needed - - Iterates until confidence threshold is met or max iterations reached -4. **Synthesis**: Generates a comprehensive research report with: +2. **Plan phase**: The research graph decomposes your question into targeted sub-questions +3. **Decision point**: Review the proposed questions in the right panel + - Add new questions using the input field + - Remove questions you don't need + - Click **Search** to execute searches for pending questions + - Click **Generate Report** to skip to synthesis (when you have enough answers) +4. **Research iterations**: After each search cycle, you return to a decision point where you can: + - Review collected answers + - Add follow-up questions based on findings + - Continue searching or generate the final report +5. **Synthesis**: Generates a comprehensive research report with: - Executive summary - Main findings with supporting evidence - Conclusions and recommendations @@ -99,32 +99,35 @@ This example demonstrates the **agent+graph** architecture pattern: - Pydantic AI agent handles user conversations - Decides when to invoke the research tool based on user intent - Responds directly to greetings/casual chat without tools - - Formats research results for the user -2. **Research Graph** (haiku.rag): +2. **Interactive Research Graph** (haiku.rag): - Multi-step research workflow invoked by the agent's tool - - Autonomous execution with plan → search → analyze → decide → synthesize flow + - Pauses at decision points waiting for human input via async queue - Emits AG-UI events for real-time progress tracking -3. **Shared Event Stream**: +3. **Decision Endpoint** (`main.py`): + - `/v1/research/decide` receives human decisions from frontend + - Forwards decisions to the waiting graph via `HumanDecision` queue + - Supports actions: `search`, `synthesize`, `modify_questions` + +4. **Shared Event Stream**: - `AGUIEmitter` is shared between agent and graph - Events from both flow through a single stream to the frontend - - Custom streaming endpoint (`main.py`) uses anyio memory streams for proper async handling + - `STATE_DELTA` events sync research state to frontend in real-time ### Components - **Backend** (Python): - Uses published `ghcr.io/ggozad/haiku.rag:latest` Docker image as base - - `agent.py`: Pydantic AI agent with `run_research` tool - - `main.py`: Custom AG-UI streaming endpoint with anyio memory object streams + - `agent.py`: Pydantic AI agent with `run_research` tool, manages `ActiveResearch` registry + - `main.py`: Custom AG-UI streaming endpoint, decision endpoint for human input - Real-time event forwarding from emitter to SSE stream - - Filters out `ACTIVITY_SNAPSHOT` events (not yet supported by CopilotKit) - **Frontend** (Next.js/React): - CopilotKit for AG-UI protocol integration - Split-pane UI: chat on left, live research state on right + - Decision UI: question editor with add/remove, search and generate report buttons - Real-time state synchronization via Server-Sent Events (SSE) - - `StateDisplay` component with collapsible sections for questions and report ## Configuration diff --git a/examples/ag-ui-research/backend/agent.py b/examples/ag-ui-research/backend/agent.py index 9537a4c3..fe15557d 100644 --- a/examples/ag-ui-research/backend/agent.py +++ b/examples/ag-ui-research/backend/agent.py @@ -1,6 +1,7 @@ """Research assistant agent with graph integration.""" -from dataclasses import dataclass +import asyncio +from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING @@ -11,7 +12,7 @@ from haiku.rag.config import load_yaml_config from haiku.rag.config.models import AppConfig from haiku.rag.graph.research.dependencies import ResearchContext from haiku.rag.graph.research.graph import build_research_graph -from haiku.rag.graph.research.state import ResearchDeps, ResearchState +from haiku.rag.graph.research.state import HumanDecision, ResearchDeps, ResearchState from haiku.rag.utils import get_model if TYPE_CHECKING: @@ -27,6 +28,20 @@ Config = ( ) +@dataclass +class ActiveResearch: + """Tracks state for active research awaiting human decision.""" + + queue: asyncio.Queue[HumanDecision] + sub_questions: list[str] = field(default_factory=list) + qa_responses: list[dict] = field(default_factory=list) + original_question: str = "" + + +# Global registry of active research by thread_id +_active_research: dict[str, ActiveResearch] = {} + + @dataclass class AgentDeps: """Dependencies for research agent.""" @@ -34,6 +49,7 @@ class AgentDeps: client: HaikuRAG agui_emitter: "AGUIEmitter[ResearchState, ResearchReport] | None" = None search_filter: str | None = None + thread_id: str | None = None model = get_model(Config.research.model, Config) @@ -50,10 +66,10 @@ CRITICAL RULES: 4. NEVER answer substantive questions from your own knowledge - always use the tool How to decide: -- "Hi" / "Hello" / "How are you?" → Respond directly, NO tools -- "What can you do?" → Respond directly, NO tools -- "How does X work in the codebase?" → Use run_research tool -- "Tell me about Y" → Use run_research tool +- "Hi" / "Hello" / "How are you?" -> Respond directly, NO tools +- "What can you do?" -> Respond directly, NO tools +- "How does X work in the codebase?" -> Use run_research tool +- "Tell me about Y" -> Use run_research tool When you use run_research, the graph will decompose questions, search the knowledge base, and generate a comprehensive report. @@ -70,23 +86,39 @@ async def run_research(ctx: RunContext[AgentDeps], question: str) -> str: DO NOT use for greetings or casual conversation. """ if ctx.deps.agui_emitter: - ctx.deps.agui_emitter.log(f"🔍 Starting research on: {question}") + ctx.deps.agui_emitter.log(f"Starting research on: {question}") - graph = build_research_graph(Config) + # Create queue for human decisions + queue: asyncio.Queue[HumanDecision] = asyncio.Queue() + + # Build interactive graph + graph = build_research_graph(Config, interactive=True) context = ResearchContext(original_question=question) state = ResearchState.from_config(context=context, config=Config) state.search_filter = ctx.deps.search_filter + # Register active research for decision endpoint to find + thread_id = ctx.deps.thread_id + if thread_id: + _active_research[thread_id] = ActiveResearch( + queue=queue, + sub_questions=[], + qa_responses=[], + original_question=question, + ) + graph_deps = ResearchDeps( client=ctx.deps.client, agui_emitter=ctx.deps.agui_emitter, + human_input_queue=queue, + interactive=True, ) try: result = await graph.run(state=state, deps=graph_deps) if ctx.deps.agui_emitter: - ctx.deps.agui_emitter.log("✅ Research complete!") + ctx.deps.agui_emitter.log("Research complete!") return f"""Research completed successfully! @@ -108,5 +140,9 @@ The full research report with all citations has been provided to the user. except Exception as e: if ctx.deps.agui_emitter: - ctx.deps.agui_emitter.log(f"❌ Research error: {str(e)}") + ctx.deps.agui_emitter.log(f"Research error: {str(e)}") return f"I encountered an error while researching: {str(e)}" + finally: + # Cleanup + if thread_id and thread_id in _active_research: + del _active_research[thread_id] diff --git a/examples/ag-ui-research/backend/main.py b/examples/ag-ui-research/backend/main.py index d432a8ba..6254c53e 100644 --- a/examples/ag-ui-research/backend/main.py +++ b/examples/ag-ui-research/backend/main.py @@ -1,8 +1,9 @@ +import json import logging import os from pathlib import Path -from agent import AgentDeps, agent +from agent import AgentDeps, _active_research, agent from anyio import create_memory_object_stream, create_task_group from anyio.streams.memory import MemoryObjectSendStream from starlette.applications import Starlette @@ -19,7 +20,7 @@ from haiku.rag.graph.agui.emitter import AGUIEmitter from haiku.rag.graph.agui.server import RunAgentInput, format_sse_event from haiku.rag.graph.research.dependencies import ResearchContext from haiku.rag.graph.research.models import ResearchReport -from haiku.rag.graph.research.state import ResearchState +from haiku.rag.graph.research.state import HumanDecision, ResearchState logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" @@ -79,11 +80,11 @@ async def stream_research_agent(request: Request) -> StreamingResponse: """Execute agent and forward emitter events to memory stream.""" async with send_stream: try: - # Create shared emitter + # Create shared emitter (use_deltas=True for CopilotKit compatibility) emitter: AGUIEmitter[ResearchState, ResearchReport] = AGUIEmitter( thread_id=input_data.thread_id, run_id=input_data.run_id, - use_deltas=False, + use_deltas=True, ) # Get client @@ -99,11 +100,14 @@ async def stream_research_agent(request: Request) -> StreamingResponse: ids_str = ", ".join(f"'{id}'" for id in document_ids) search_filter = f"id IN ({ids_str})" + thread_id = input_data.thread_id + # Create agent dependencies with shared emitter agent_deps = AgentDeps( client=client, agui_emitter=emitter, search_filter=search_filter, + thread_id=thread_id, ) # Start run with empty initial state @@ -148,6 +152,37 @@ async def stream_research_agent(request: Request) -> StreamingResponse: await send_stream.send(format_sse_event(delta_event)) continue + # When human_decision tool starts, set awaiting_decision flag + if event_type == "TOOL_CALL_START": + tool_name = event.get("toolCallName") + if tool_name == "human_decision": + delta_event = { + "type": "STATE_DELTA", + "delta": [ + { + "op": "add", + "path": "/awaiting_decision", + "value": True, + } + ], + } + await send_stream.send(format_sse_event(delta_event)) + + # Sync state to ActiveResearch when human_decision tool call + if event_type == "TOOL_CALL_ARGS" and thread_id: + delta = event.get("delta", "{}") + args = ( + json.loads(delta) if isinstance(delta, str) else delta + ) + active = _active_research.get(thread_id) + if active: + active.sub_questions = list( + args.get("sub_questions", []) + ) + active.qa_responses = list(args.get("qa_responses", [])) + if "original_question" in args: + active.original_question = args["original_question"] + await send_stream.send(format_sse_event(event)) # Run agent and event forwarding concurrently @@ -248,10 +283,36 @@ async def visualize_chunk(request: Request) -> JSONResponse: ) +async def research_decide(request: Request) -> JSONResponse: + """Endpoint to receive human decisions for active research.""" + body = await request.json() + action = body.get("action", "search") + questions = body.get("questions", []) + + # Get first active research (single-user example) + active = next(iter(_active_research.values()), None) + if not active: + return JSONResponse({"error": "No active research found"}, status_code=404) + + # When "search" action is sent with questions, use "modify_questions" to update them + effective_action = ( + "modify_questions" if action == "search" and questions else action + ) + + decision = HumanDecision( + action=effective_action, + questions=questions or None, + ) + await active.queue.put(decision) + + return JSONResponse({"status": "ok", "action": effective_action}) + + # Create Starlette app app = Starlette( routes=[ Route("/v1/research/stream", stream_research_agent, methods=["POST"]), + Route("/v1/research/decide", research_decide, methods=["POST"]), Route("/api/documents", list_documents, methods=["GET"]), Route("/api/visualize/{chunk_id}", visualize_chunk, methods=["GET"]), Route("/health", health_check, methods=["GET"]), diff --git a/examples/ag-ui-research/frontend/components/Agent.tsx b/examples/ag-ui-research/frontend/components/Agent.tsx index 75429a91..4cc1c731 100644 --- a/examples/ag-ui-research/frontend/components/Agent.tsx +++ b/examples/ag-ui-research/frontend/components/Agent.tsx @@ -3,6 +3,7 @@ import { CopilotKit, useCoAgent } from "@copilotkit/react-core"; import { CopilotChat } from "@copilotkit/react-ui"; import "@copilotkit/react-ui/styles.css"; +import { useState, useEffect } from "react"; import DocumentSelector from "./DocumentSelector"; import StateDisplay from "./StateDisplay"; @@ -58,6 +59,7 @@ interface ResearchState { current_activity?: string; current_activity_message?: string; documentFilter?: string[]; + awaiting_decision?: boolean; } function AgentContent() { @@ -75,13 +77,61 @@ function AgentContent() { max_concurrency: 1, last_eval: null, documentFilter: [], + awaiting_decision: false, }, }); + const [editableQuestions, setEditableQuestions] = useState([]); + const [newQuestion, setNewQuestion] = useState(""); + const [submitting, setSubmitting] = useState(false); + + // Sync editable questions when state changes + useEffect(() => { + if (state.awaiting_decision && state.context.sub_questions) { + setEditableQuestions([...state.context.sub_questions]); + } + }, [state.awaiting_decision, state.context.sub_questions]); + const handleDocumentFilterChange = (ids: string[]) => { setState({ ...state, documentFilter: ids }); }; + const handleRemoveQuestion = (index: number) => { + setEditableQuestions(editableQuestions.filter((_, i) => i !== index)); + }; + + const handleAddQuestion = () => { + if (newQuestion.trim()) { + setEditableQuestions([...editableQuestions, newQuestion.trim()]); + setNewQuestion(""); + } + }; + + const handleDecision = async (action: "search" | "synthesize") => { + setSubmitting(true); + try { + const response = await fetch( + `${process.env.NEXT_PUBLIC_API_URL || "http://localhost:8000"}/v1/research/decide`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + thread_id: state.context.original_question, // Use as identifier + action, + questions: editableQuestions, + }), + } + ); + if (response.ok) { + setState({ ...state, awaiting_decision: false }); + } + } catch (error) { + console.error("Failed to send decision:", error); + } finally { + setSubmitting(false); + } + }; + return ( <>