Upload 3 files

app/agent_system.py

@@ -0,0 +1,360 @@
+import os
+import json
+import asyncio
+import traceback
+from openai import AsyncOpenAI
+from typing import AsyncGenerator
+from docs_context import PRAISONAI_DOCS
+
+LONGCAT_BASE_URL = "https://api.longcat.chat/openai/v1"
+MODEL = "LongCat-Flash-Lite"
+
+
+def build_orchestrator_system():
+    return f"""You are the Main Orchestrator Agent for PraisonChat — a powerful AI system that solves complex tasks by dynamically creating specialized sub-agents, each with custom-built tools.
+
+{PRAISONAI_DOCS}
+
+## Your Job
+When a user sends a task:
+1. Analyze what kind of work is needed
+2. Design specialized sub-agents, each focused on one responsibility
+3. For each sub-agent, design the exact Python tools they need
+4. Return a structured execution plan as JSON
+
+## Response Format
+Always respond with valid JSON in this exact structure:
+{{
+  "task_analysis": "Clear explanation of what needs to be done and why",
+  "needs_sub_agents": true,
+  "sub_agents": [
+    {{
+      "name": "AgentName",
+      "role": "Specific professional role",
+      "goal": "What this agent achieves",
+      "backstory": "Brief agent background/expertise",
+      "tools": [
+        {{
+          "name": "tool_function_name",
+          "description": "What this tool does",
+          "parameters": "param1: str, param2: int = 10",
+          "return_type": "str",
+          "docstring": "Detailed docstring explaining the tool",
+          "implementation": "Python code body (indented with 4 spaces, no def line)"
+        }}
+      ],
+      "task_description": "Exact task for this agent to perform",
+      "expected_output": "What format/content to expect back"
+    }}
+  ],
+  "execution_order": ["AgentName1", "AgentName2"],
+  "synthesis_instruction": "How to combine all agent results into the final answer"
+}}
+
+## Rules
+- Create sub-agents ONLY when needed. Simple questions = no sub-agents (set needs_sub_agents: false)
+- Tools must be real, executable Python code
+- Each tool implementation must be complete and working
+- Maximum 4 sub-agents per task
+- Tool implementations must not import anything not in Python stdlib
+- Keep tool implementations under 30 lines each
+"""
+
+
+def build_tool_function(tool_spec: dict) -> callable:
+    """Dynamically create a Python function from a tool spec."""
+    name = tool_spec["name"]
+    params = tool_spec.get("parameters", "input: str")
+    return_type = tool_spec.get("return_type", "str")
+    docstring = tool_spec.get("docstring", "Auto-generated tool")
+    implementation = tool_spec.get("implementation", "return str(input)")
+
+    # Build the function source
+    func_source = f"""
+def {name}({params}) -> {return_type}:
+    \"\"\"{docstring}\"\"\"
+    {chr(10).join('    ' + line if line.strip() else '' for line in implementation.strip().splitlines())}
+"""
+    namespace = {}
+    try:
+        exec(func_source, namespace)
+        return namespace[name]
+    except Exception as e:
+        # Return a safe fallback function
+        def fallback_tool(**kwargs) -> str:
+            return f"Tool '{name}' could not be created: {e}. Using fallback."
+        fallback_tool.__name__ = name
+        fallback_tool.__doc__ = docstring
+        return fallback_tool
+
+
+class AgentOrchestrator:
+    def __init__(self):
+        self._client_cache = {}
+
+    def get_client(self, api_key: str) -> AsyncOpenAI:
+        if api_key not in self._client_cache:
+            self._client_cache[api_key] = AsyncOpenAI(
+                api_key=api_key,
+                base_url=LONGCAT_BASE_URL
+            )
+        return self._client_cache[api_key]
+
+    async def plan_task(self, client: AsyncOpenAI, user_message: str, history: list) -> dict:
+        """Ask the orchestrator to plan the task."""
+        messages = [{"role": "system", "content": build_orchestrator_system()}]
+
+        # Add recent history for context
+        for msg in history[-6:]:
+            messages.append({"role": msg["role"], "content": str(msg["content"])[:2000]})
+
+        messages.append({
+            "role": "user",
+            "content": f"Plan the execution for this task: {user_message}"
+        })
+
+        response = await client.chat.completions.create(
+            model=MODEL,
+            messages=messages,
+            max_tokens=6000,
+            temperature=0.2,
+        )
+
+        raw = response.choices[0].message.content.strip()
+        # Strip markdown code fences if present
+        if raw.startswith("```"):
+            raw = raw.split("```")[1]
+            if raw.startswith("json"):
+                raw = raw[4:]
+        raw = raw.strip()
+
+        try:
+            return json.loads(raw)
+        except Exception:
+            return {
+                "task_analysis": "Direct response",
+                "needs_sub_agents": False,
+                "sub_agents": [],
+                "execution_order": [],
+                "synthesis_instruction": "Respond directly"
+            }
+
+    async def run_sub_agent(
+        self,
+        client: AsyncOpenAI,
+        agent_spec: dict,
+        context_so_far: str
+    ) -> str:
+        """Run a single sub-agent with its tools."""
+        tool_descriptions = ""
+        tools_created = []
+        tool_errors = []
+
+        for tool_spec in agent_spec.get("tools", []):
+            fn = build_tool_function(tool_spec)
+            tools_created.append(fn.__name__)
+            tool_descriptions += f"\n- {fn.__name__}: {tool_spec.get('description', '')}"
+
+        system_prompt = f"""You are {agent_spec['name']}.
+Role: {agent_spec['role']}
+Goal: {agent_spec['goal']}
+Backstory: {agent_spec.get('backstory', '')}
+
+You have access to these custom tools (simulate their usage in your reasoning):
+{tool_descriptions if tool_descriptions else "No specialized tools - use your knowledge directly."}
+
+Context from previous agents:
+{context_so_far if context_so_far else "You are the first agent running."}
+
+Execute your task thoroughly. Show your reasoning and tool usage step by step.
+Expected output: {agent_spec.get('expected_output', 'Detailed results')}"""
+
+        response = await client.chat.completions.create(
+            model=MODEL,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": agent_spec["task_description"]}
+            ],
+            max_tokens=12000,
+            temperature=0.7,
+        )
+
+        return response.choices[0].message.content
+
+    async def synthesize(
+        self,
+        client: AsyncOpenAI,
+        user_message: str,
+        agent_results: dict,
+        synthesis_instruction: str
+    ) -> AsyncGenerator[str, None]:
+        """Stream the final synthesized response."""
+        results_text = "\n\n".join([
+            f"=== {name} ===\n{result}"
+            for name, result in agent_results.items()
+        ])
+
+        system_prompt = f"""You are the Main Orchestrator synthesizing results from specialized sub-agents.
+
+Synthesis instruction: {synthesis_instruction}
+
+Sub-agent results:
+{results_text}
+
+Provide a comprehensive, well-structured final response to the user.
+Use markdown formatting. Be thorough but concise."""
+
+        stream = await client.chat.completions.create(
+            model=MODEL,
+            messages=[
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_message}
+            ],
+            max_tokens=16000,
+            temperature=0.7,
+            stream=True
+        )
+
+        async for chunk in stream:
+            delta = chunk.choices[0].delta
+            if delta.content:
+                yield delta.content
+
+    async def direct_response(
+        self,
+        client: AsyncOpenAI,
+        user_message: str,
+        history: list
+    ) -> AsyncGenerator[str, None]:
+        """Stream a direct response without sub-agents."""
+        messages = [{
+            "role": "system",
+            "content": "You are PraisonChat, a powerful AI assistant. Respond helpfully using markdown formatting."
+        }]
+        for msg in history[-10:]:
+            messages.append({"role": msg["role"], "content": str(msg["content"])[:3000]})
+        messages.append({"role": "user", "content": user_message})
+
+        stream = await client.chat.completions.create(
+            model=MODEL,
+            messages=messages,
+            max_tokens=16000,
+            temperature=0.7,
+            stream=True
+        )
+
+        async for chunk in stream:
+            delta = chunk.choices[0].delta
+            if delta.content:
+                yield delta.content
+
+    async def stream_response(
+        self,
+        user_message: str,
+        history: list,
+        api_key: str
+    ) -> AsyncGenerator[str, None]:
+        """Main entry point — streams SSE-formatted events."""
+
+        def emit(data: dict) -> str:
+            return json.dumps(data)
+
+        client = self.get_client(api_key)
+
+        try:
+            # ── Step 1: Plan ─────────────────────────────────────────────────
+            yield emit({"type": "step", "text": "🧠 Main Agent analyzing your task..."})
+            await asyncio.sleep(0)
+
+            plan = await self.plan_task(client, user_message, history)
+
+            yield emit({
+                "type": "step",
+                "text": f"📋 {plan.get('task_analysis', 'Planning execution...')}"
+            })
+            await asyncio.sleep(0)
+
+            sub_agents = plan.get("sub_agents", [])
+            needs_sub_agents = plan.get("needs_sub_agents", bool(sub_agents))
+
+            # ── Step 2: Sub-agents or direct ─────────────────────────────────
+            if needs_sub_agents and sub_agents:
+                yield emit({
+                    "type": "step",
+                    "text": f"🤖 Spawning {len(sub_agents)} specialized sub-agent(s)..."
+                })
+
+                for agent_spec in sub_agents:
+                    tool_names = [t["name"] for t in agent_spec.get("tools", [])]
+                    yield emit({
+                        "type": "agent_created",
+                        "name": agent_spec["name"],
+                        "role": agent_spec["role"],
+                        "tools": tool_names
+                    })
+                    await asyncio.sleep(0.05)
+
+                # Execute each sub-agent
+                context_so_far = ""
+                agent_results = {}
+                execution_order = plan.get("execution_order", [a["name"] for a in sub_agents])
+
+                for agent_name in execution_order:
+                    agent_spec = next(
+                        (a for a in sub_agents if a["name"] == agent_name), None
+                    )
+                    if not agent_spec:
+                        continue
+
+                    yield emit({
+                        "type": "step",
+                        "text": f"⚡ {agent_name} working on: {agent_spec['task_description'][:100]}..."
+                    })
+                    await asyncio.sleep(0)
+
+                    try:
+                        result = await self.run_sub_agent(client, agent_spec, context_so_far)
+                        agent_results[agent_name] = result
+                        context_so_far += f"\n\n{agent_name} completed: {result[:600]}"
+
+                        yield emit({
+                            "type": "agent_result",
+                            "name": agent_name,
+                            "preview": result[:300] + ("..." if len(result) > 300 else "")
+                        })
+                    except Exception as e:
+                        yield emit({
+                            "type": "step",
+                            "text": f"⚠️ {agent_name} encountered an issue: {str(e)[:100]}"
+                        })
+                        agent_results[agent_name] = f"Error: {e}"
+
+                # Synthesize
+                yield emit({"type": "step", "text": "✨ Synthesizing final response..."})
+                yield emit({"type": "response_start"})
+                await asyncio.sleep(0)
+
+                async for token in self.synthesize(
+                    client, user_message, agent_results,
+                    plan.get("synthesis_instruction", "Combine all results into a clear response")
+                ):
+                    yield emit({"type": "token", "content": token})
+
+            else:
+                # Direct response
+                yield emit({"type": "step", "text": "💬 Generating response..."})
+                yield emit({"type": "response_start"})
+                await asyncio.sleep(0)
+
+                async for token in self.direct_response(client, user_message, history):
+                    yield emit({"type": "token", "content": token})
+
+            yield emit({"type": "done"})
+
+        except Exception as e:
+            tb = traceback.format_exc()
+            yield emit({"type": "error", "message": str(e), "detail": tb[:500]})
+
+
+# Singleton
+orchestrator = AgentOrchestrator()

app/docs_context.py

@@ -0,0 +1,188 @@
+PRAISONAI_DOCS = """
+# PraisonAI Agent Framework — Full Documentation
+
+## What is PraisonAI?
+PraisonAI is a multi-agent orchestration framework. You (the orchestrator) can create specialized
+sub-agents, each with custom tools written in Python, to complete complex tasks collaboratively.
+
+---
+
+## Creating an Agent
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+agent = Agent(
+    name="ResearchAgent",           # Unique agent name
+    role="Research Specialist",     # Role description
+    goal="Find and summarize info", # What the agent aims to achieve
+    backstory="Expert researcher",  # Context/personality
+    llm="LongCat-Flash-Lite",       # LLM to use
+    tools=[my_tool_fn],             # List of tool functions
+    verbose=True,
+    self_reflect=False,             # Enable self-reflection
+    min_reflect=1,
+    max_reflect=3
+)
+```
+
+---
+
+## Creating a Task
+```python
+task = Task(
+    name="research_task",
+    description="Research the latest developments in quantum computing",
+    expected_output="A detailed summary with key findings",
+    agent=agent,
+    context=[previous_task],   # Task dependencies
+    tools=[tool_fn]            # Override agent tools
+)
+```
+
+---
+
+## Creating Custom Tools
+Tools are just Python functions with type hints and docstrings:
+
+```python
+def search_web(query: str) -> str:
+    """Search the internet for information.
+    Args:
+        query: The search query string
+    Returns:
+        Search results as a formatted string
+    """
+    import requests
+    # implementation
+    return results
+
+def read_file(filepath: str) -> str:
+    """Read content from a file.
+    Args:
+        filepath: Path to the file
+    Returns:
+        File content as string
+    """
+    with open(filepath) as f:
+        return f.read()
+
+def execute_python(code: str) -> str:
+    """Execute Python code and return output.
+    Args:
+        code: Python code to execute
+    Returns:
+        Execution output
+    """
+    import subprocess
+    result = subprocess.run(['python3', '-c', code], capture_output=True, text=True)
+    return result.stdout + result.stderr
+
+def calculate(expression: str) -> str:
+    """Evaluate a mathematical expression.
+    Args:
+        expression: Math expression as string
+    Returns:
+        Result as string
+    """
+    return str(eval(expression))
+```
+
+---
+
+## Running Multiple Agents (Sequential)
+```python
+system = PraisonAIAgents(
+    agents=[researcher, analyst, writer],
+    tasks=[research_task, analysis_task, write_task],
+    process="sequential",   # agents run one after another
+    verbose=True
+)
+result = system.start()
+```
+
+---
+
+## Running Multiple Agents (Hierarchical)
+```python
+# Manager coordinates workers automatically
+system = PraisonAIAgents(
+    agents=[manager_agent, worker1, worker2],
+    tasks=[main_task, sub_task1, sub_task2],
+    process="hierarchical",
+    manager_llm="LongCat-Flash-Lite",
+    verbose=True
+)
+result = system.start()
+```
+
+---
+
+## Task Context (Chaining Tasks)
+```python
+# task2 receives output of task1 as context
+task1 = Task(name="t1", description="...", agent=agent1, expected_output="...")
+task2 = Task(name="t2", description="...", agent=agent2, expected_output="...", context=[task1])
+```
+
+---
+
+## Best Practices for Dynamic Tool Creation
+When you need a tool for a specific purpose, write it as a Python function:
+
+1. **Naming**: Use snake_case, descriptive names (e.g., `fetch_github_repo`, `parse_csv_data`)
+2. **Type hints**: Always include type hints for parameters and return value
+3. **Docstrings**: Write clear docstrings — this is how the agent understands the tool
+4. **Error handling**: Wrap risky operations in try/except
+5. **Return strings**: Tools should return strings for easy consumption
+
+Example of a dynamically created specialized tool:
+```python
+def analyze_sentiment(text: str) -> str:
+    \"\"\"Analyze the sentiment of given text.
+    Args:
+        text: The text to analyze
+    Returns:
+        Sentiment analysis result: positive, negative, or neutral with confidence
+    \"\"\"
+    # Simple keyword-based analysis
+    positive_words = ['good', 'great', 'excellent', 'happy', 'love', 'best']
+    negative_words = ['bad', 'terrible', 'hate', 'worst', 'awful', 'horrible']
+    text_lower = text.lower()
+    pos = sum(1 for w in positive_words if w in text_lower)
+    neg = sum(1 for w in negative_words if w in text_lower)
+    if pos > neg:
+        return f"Positive sentiment (score: {pos}/{pos+neg})"
+    elif neg > pos:
+        return f"Negative sentiment (score: {neg}/{pos+neg})"
+    else:
+        return "Neutral sentiment"
+```
+
+---
+
+## Sub-Agent Design Patterns
+
+### Pattern 1: Research + Synthesis
+- Agent 1 (Researcher): Gathers raw information using search/fetch tools
+- Agent 2 (Analyst): Processes and analyzes the gathered data
+- Agent 3 (Writer): Produces the final output
+
+### Pattern 2: Divide and Conquer
+- Split large tasks into parallel sub-tasks
+- Each agent handles one domain
+- Orchestrator merges results
+
+### Pattern 3: Validation Pipeline
+- Agent 1: Performs the main task
+- Agent 2: Reviews and validates Agent 1's output
+- Agent 3: Applies corrections
+
+---
+
+## LongCat-Flash-Lite Configuration
+Model: LongCat-Flash-Lite
+Max tokens: 327,680 (320K context)
+Speed: Very fast (~500-700 tokens/sec)
+Free quota: 50M tokens/day
+API Base: https://api.longcat.chat/openai/v1
+"""

app/main.py

@@ -0,0 +1,67 @@
+import os
+import json
+from pathlib import Path
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+from agent_system import orchestrator
+
+app = FastAPI(title="PraisonChat", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+STATIC_DIR = Path(__file__).parent / "static"
+
+
+@app.get("/", response_class=HTMLResponse)
+async def root():
+    return HTMLResponse(content=(STATIC_DIR / "index.html").read_text(encoding="utf-8"))
+
+
+@app.get("/api/health")
+def health():
+    return {"status": "ok", "model": "LongCat-Flash-Lite", "version": "1.0.0"}
+
+
+@app.get("/api/models")
+def models():
+    return {
+        "models": [
+            {"id": "LongCat-Flash-Lite",         "name": "LongCat Flash Lite",         "context": "320K", "speed": "fastest"},
+            {"id": "LongCat-Flash-Chat",          "name": "LongCat Flash Chat",         "context": "256K", "speed": "fast"},
+            {"id": "LongCat-Flash-Thinking-2601", "name": "LongCat Flash Thinking",     "context": "256K", "speed": "medium"},
+        ]
+    }
+
+
+@app.post("/api/chat")
+async def chat(request: Request):
+    try:
+        body = await request.json()
+    except Exception:
+        raise HTTPException(400, "Invalid JSON body")
+
+    messages  = body.get("messages", [])
+    api_key   = body.get("api_key") or os.getenv("LONGCAT_API_KEY", "")
+
+    if not api_key:
+        raise HTTPException(400, "LongCat API key is required. Set it in Settings or as LONGCAT_API_KEY env var.")
+
+    if not messages:
+        raise HTTPException(400, "No messages provided")
+
+    user_message = messages[-1].get("content", "")
+    history      = messages[:-1]
+
+    async def event_stream():
+        async for chunk in orchestrator.stream_response(user_message, history, api_key):
+            yield f"data: {chunk}\n\n"
+
+    return StreamingResponse(event_stream(), media_type="text/event-stream",
+                             headers={"X-Accel-Buffering": "no", "Cache-Control": "no-cache"})