Richard/simple agent

backend/onyx/chat/process_message.py

@@ -3,13 +3,16 @@
 import traceback
 from collections.abc import Callable
 from collections.abc import Iterator
+from typing import Any
 from typing import cast
+from typing import Dict
 from typing import Protocol
 
 from sqlalchemy.orm import Session
 
 from onyx.agents.agent_search.orchestration.nodes.call_tool import ToolCallException
 from onyx.chat.answer import Answer
+from onyx.chat.answer_scratchpad import stream_chat_sync
 from onyx.chat.chat_utils import create_chat_chain
 from onyx.chat.chat_utils import create_temporary_persona
 from onyx.chat.chat_utils import process_kg_commands
@@ -24,9 +27,6 @@
 from onyx.chat.models import QADocsResponse
 from onyx.chat.models import StreamingError
 from onyx.chat.models import UserKnowledgeFilePacket
-from onyx.chat.packet_proccessing.process_streamed_packets import (
-    process_streamed_packets,
-)
 from onyx.chat.prompt_builder.answer_prompt_builder import AnswerPromptBuilder
 from onyx.chat.prompt_builder.answer_prompt_builder import default_build_system_message
 from onyx.chat.prompt_builder.answer_prompt_builder import default_build_user_message
@@ -41,7 +41,6 @@
 from onyx.context.search.enums import OptionalSearchSetting
 from onyx.context.search.models import InferenceSection
 from onyx.context.search.models import RetrievalDetails
-from onyx.context.search.models import SavedSearchDoc
 from onyx.context.search.retrieval.search_runner import (
     inference_sections_from_ids,
 )
@@ -76,11 +75,7 @@
 from onyx.llm.utils import litellm_exception_to_error_msg
 from onyx.natural_language_processing.utils import get_tokenizer
 from onyx.server.query_and_chat.models import CreateChatMessageRequest
-from onyx.server.query_and_chat.streaming_models import CitationDelta
 from onyx.server.query_and_chat.streaming_models import CitationInfo
-from onyx.server.query_and_chat.streaming_models import MessageDelta
-from onyx.server.query_and_chat.streaming_models import MessageStart
-from onyx.server.query_and_chat.streaming_models import Packet
 from onyx.server.utils import get_json_line
 from onyx.tools.force import ForceUseTool
 from onyx.tools.models import SearchToolOverrideKwargs
@@ -671,10 +666,49 @@ def stream_chat_message_objects(
             use_agentic_search=new_msg_req.use_agentic_search,
             skip_gen_ai_answer_generation=new_msg_req.skip_gen_ai_answer_generation,
         )
-
-        # Process streamed packets using the new packet processing module
-        yield from process_streamed_packets(
-            answer_processed_output=answer.processed_streamed_output,
+        type_to_role = {
+            "human": "user",
+            "assistant": "assistant",
+            "system": "system",
+            "function": "function",
+        }
+        SYSTEM_PROMPT = """
+        You are a highly capable, thoughtful, and precise assistant. Your goal is to deeply understand the \
+        user's intent, ask clarifying questions when needed, think step-by-step through complex problems, \
+        provide clear and accurate answers, and proactively anticipate helpful follow-up information. Always \
+        prioritize being truthful, nuanced, insightful, and efficient.
+        The current date is September 18, 2025.
+
+        You use different text styles, bolding, emojis (sparingly), block quotes, and other formatting to make \
+        your responses more readable and engaging.
+        You use proper Markdown and LaTeX to format your responses for math, scientific, and chemical formulas, \
+        symbols, etc.: '$$\\n[expression]\\n$$' for standalone cases and '\\( [expression] \\)' when inline.
+        For code you prefer to use Markdown and specify the language.
+        You can use Markdown horizontal rules (---) to separate sections of your responses.
+        You can use Markdown tables to format your responses for data, lists, and other structured information.
+
+        You must cite inline using tags from tool results.
+
+        Rules:
+        - Only cite sources provided by the tools (use each item’s "tag" field).
+        - Place the citation immediately after the claim it supports, like this: "... result [S1](https://linkforS1)" or
+         "... results [S1](https://linkforS1)[S3](https://linkforS3)".
+        - If multiple sentences in a row are supported by the same source, cite the first sentence;
+        then omit repeats until the source changes.
+        - Never invent tags. If no source supports a claim, say so.
+        - Do not add a separate “Sources” section unless asked.
+        """
+        system_message = [{"role": "system", "content": SYSTEM_PROMPT}]
+        other_messages = [
+            {"role": type_to_role[message.type], "content": message.content}
+            for message in answer.graph_inputs.prompt_builder.build()
+            if message.type != "system"
+        ]
+        yield from stream_chat_sync(
+            messages=system_message + other_messages,
+            cfg=answer.graph_config,
+            llm=answer.graph_tooling.primary_llm,
+            search_tool=answer.graph_tooling.search_tool,
         )
 
     except ValueError as e:
@@ -736,7 +770,15 @@ def stream_chat_message(
                 document_retrieval_latency = time.time() - start_time
                 logger.debug(f"First doc time: {document_retrieval_latency}")
 
-            yield get_json_line(obj.model_dump())
+            # Convert Pydantic models to dictionaries for JSON serialization
+            if hasattr(obj, "model_dump"):
+                obj_dict = obj.model_dump()
+            elif hasattr(obj, "dict"):
+                obj_dict = obj.dict()
+            else:
+                obj_dict = obj
+
+            yield get_json_line(obj_dict)
 
 
 def remove_answer_citations(answer: str) -> str:
@@ -745,48 +787,98 @@ def remove_answer_citations(answer: str) -> str:
     return re.sub(pattern, "", answer)
 
 
+def _convert_to_packet_obj(packet: Dict[str, Any]) -> Any | None:
+    """Convert a packet dictionary to PacketObj when possible.
+
+    Args:
+        packet: Dictionary containing packet data
+
+    Returns:
+        PacketObj instance if conversion is possible, None otherwise
+    """
+    if not isinstance(packet, dict) or "type" not in packet:
+        return None
+
+    packet_type = packet.get("type")
+    if not packet_type:
+        return None
+
+    try:
+        # Import here to avoid circular imports
+        from onyx.server.query_and_chat.streaming_models import (
+            MessageStart,
+            MessageDelta,
+            OverallStop,
+            SectionEnd,
+            SearchToolStart,
+            SearchToolDelta,
+            ImageGenerationToolStart,
+            ImageGenerationToolDelta,
+            ImageGenerationToolHeartbeat,
+            CustomToolStart,
+            CustomToolDelta,
+            ReasoningStart,
+            ReasoningDelta,
+            CitationStart,
+            CitationDelta,
+        )
+
+        # Map packet types to their corresponding classes
+        type_mapping = {
+            "message_start": MessageStart,
+            "message_delta": MessageDelta,
+            "stop": OverallStop,
+            "section_end": SectionEnd,
+            "internal_search_tool_start": SearchToolStart,
+            "internal_search_tool_delta": SearchToolDelta,
+            "image_generation_tool_start": ImageGenerationToolStart,
+            "image_generation_tool_delta": ImageGenerationToolDelta,
+            "image_generation_tool_heartbeat": ImageGenerationToolHeartbeat,
+            "custom_tool_start": CustomToolStart,
+            "custom_tool_delta": CustomToolDelta,
+            "reasoning_start": ReasoningStart,
+            "reasoning_delta": ReasoningDelta,
+            "citation_start": CitationStart,
+            "citation_delta": CitationDelta,
+        }
+
+        packet_class = type_mapping.get(packet_type)
+        if packet_class:
+            # Create instance using the packet data, filtering out None values
+            filtered_data = {k: v for k, v in packet.items() if v is not None}
+            return packet_class(**filtered_data)
+
+    except Exception as e:
+        # Log the error but don't fail the entire process
+        logger.debug(f"Failed to convert packet to PacketObj: {e}")
+
+    return None
+
+
 @log_function_time()
 def gather_stream(
-    packets: AnswerStream,
+    packets: Iterator[Dict[str, Any]],
 ) -> ChatBasicResponse:
     answer = ""
-    citations: list[CitationInfo] = []
-    error_msg: str | None = None
-    message_id: int | None = None
-    top_documents: list[SavedSearchDoc] = []
-
     for packet in packets:
-        if isinstance(packet, Packet):
-            # Handle the different packet object types
-            if isinstance(packet.obj, MessageStart):
-                # MessageStart contains the initial content and final documents
-                if packet.obj.content:
-                    answer += packet.obj.content
-                if packet.obj.final_documents:
-                    top_documents = packet.obj.final_documents
-            elif isinstance(packet.obj, MessageDelta):
-                # MessageDelta contains incremental content updates
-                if packet.obj.content:
-                    answer += packet.obj.content
-            elif isinstance(packet.obj, CitationDelta):
-                # CitationDelta contains citation information
-                if packet.obj.citations:
-                    citations.extend(packet.obj.citations)
-        elif isinstance(packet, StreamingError):
-            error_msg = packet.error
-        elif isinstance(packet, MessageResponseIDInfo):
-            message_id = packet.reserved_assistant_message_id
-
-    if message_id is None:
-        raise ValueError("Message ID is required")
+        if packet != {"type": "event"}:
+            print(packet)
+
+        # Convert packet to PacketObj when possible
+        packet_obj = _convert_to_packet_obj(packet)
+        if packet_obj:
+            # Handle PacketObj types that contain text content
+            if hasattr(packet_obj, "content") and packet_obj.content:
+                answer += packet_obj.content
+        elif "text" in packet:
+            # Fallback for legacy packet format
+            answer += packet["text"]
 
     return ChatBasicResponse(
         answer=answer,
         answer_citationless=remove_answer_citations(answer),
-        cited_documents={
-            citation.citation_num: citation.document_id for citation in citations
-        },
-        message_id=message_id,
-        error_msg=error_msg,
-        top_documents=top_documents,
+        cited_documents={},
+        message_id=0,
+        error_msg=None,
+        top_documents=[],
     )

backend/onyx/evals/demo_agent.py

@@ -0,0 +1,79 @@
+import asyncio
+import os
+
+from agents import ModelSettings
+from agents import run_demo_loop
+from agents.agent import Agent
+from agents.extensions.handoff_prompt import prompt_with_handoff_instructions
+from agents.extensions.models.litellm_model import LitellmModel
+from pydantic import BaseModel
+
+from onyx.agents.agent_search.dr.dr_prompt_builder import (
+    get_dr_prompt_orchestration_templates,
+)
+from onyx.agents.agent_search.dr.enums import ResearchType
+from onyx.agents.agent_search.dr.models import DRPromptPurpose
+
+
+def construct_simple_agent() -> Agent:
+    litellm_model = LitellmModel(
+        model="gpt-4.1",
+        api_key=os.getenv("OPENAI_API_KEY"),
+    )
+    return Agent(
+        name="Assistant",
+        instructions="""
+        You are a helpful assistant that can search the web, fetch content from URLs,
+        and search internal databases.
+        """,
+        model=litellm_model,
+        tools=[],
+        model_settings=ModelSettings(
+            temperature=0.0,
+            include_usage=True,  # Track usage metrics
+        ),
+    )
+
+
+class ClarificationOutput(BaseModel):
+    clarification_question: str
+    clarification_needed: bool
+
+
+def construct_dr_agent() -> Agent:
+    simple_agent = construct_simple_agent()
+    litellm_model = LitellmModel(
+        model="gpt-4.1",
+        api_key=os.getenv("OPENAI_API_KEY"),
+    )
+    base_clarification_prompt = get_dr_prompt_orchestration_templates(
+        DRPromptPurpose.CLARIFICATION,
+        research_type=ResearchType.DEEP,
+        entity_types_string=None,
+        relationship_types_string=None,
+        available_tools={},
+    )
+    clarification_prompt = base_clarification_prompt.build(
+        question="",
+        chat_history_string="",
+    )
+    clarifier_prompt = prompt_with_handoff_instructions(clarification_prompt)
+    clarifier_agent = Agent(
+        name="Clarifier",
+        instructions=clarifier_prompt,
+        model=litellm_model,
+        tools=[],
+        output_type=ClarificationOutput,
+        handoffs=[simple_agent],
+        model_settings=ModelSettings(tool_choice="required"),
+    )
+    return clarifier_agent
+
+
+async def main() -> None:
+    agent = construct_dr_agent()
+    await run_demo_loop(agent)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

backend/onyx/evals/eval.py

@@ -100,6 +100,7 @@ def run_eval(
     data: list[dict[str, dict[str, str]]] | None = None,
     remote_dataset_name: str | None = None,
     provider: EvalProvider = get_default_provider(),
+    no_send_logs: bool = False,
 ) -> EvalationAck:
     if data is not None and remote_dataset_name is not None:
         raise ValueError("Cannot specify both data and remote_dataset_name")
@@ -112,4 +113,5 @@ def run_eval(
         configuration=configuration,
         data=data,
         remote_dataset_name=remote_dataset_name,
+        no_send_logs=no_send_logs,
     )

backend/onyx/evals/eval_cli.py

@@ -42,6 +42,7 @@ def run_local(
     local_data_path: str | None,
     remote_dataset_name: str | None,
     search_permissions_email: str | None = None,
+    no_send_logs: bool = False,
 ) -> EvalationAck:
     """
     Run evaluation with local configurations.
@@ -67,15 +68,19 @@ def run_local(
 
     if remote_dataset_name:
         score = run_eval(
-            configuration=configuration, remote_dataset_name=remote_dataset_name
+            configuration=configuration,
+            remote_dataset_name=remote_dataset_name,
+            no_send_logs=no_send_logs,
         )
     else:
         if local_data_path is None:
             raise ValueError(
                 "local_data_path or remote_dataset_name is required for local evaluation"
             )
         data = load_data_local(local_data_path)
-        score = run_eval(configuration=configuration, data=data)
+        score = run_eval(
+            configuration=configuration, data=data, no_send_logs=no_send_logs
+        )
 
     return score
 
@@ -172,6 +177,13 @@ def main() -> None:
         help="Email address to impersonate for the evaluation",
     )
 
+    parser.add_argument(
+        "--no-send-logs",
+        action="store_true",
+        help="Do not send logs to the remote server",
+        default=False,
+    )
+
     args = parser.parse_args()
 
     if args.local_data_path:
@@ -215,6 +227,7 @@ def main() -> None:
             local_data_path=args.local_data_path,
             remote_dataset_name=args.remote_dataset_name,
             search_permissions_email=args.search_permissions_email,
+            no_send_logs=args.no_send_logs,
         )