feat: initial release - Hunzi agent framework v0.1.0
This commit is contained in:
@@ -0,0 +1,336 @@
|
||||
"""工作流管理 API"""
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from fastapi import APIRouter, HTTPException, BackgroundTasks
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.database import get_session
|
||||
from app.models import Workflow, WorkflowRun, WorkflowStep
|
||||
from app.workflow.models import Workflow as WorkflowModel, StepDef
|
||||
from app.workflow.engine import WorkflowEngine
|
||||
from app.config import get_config
|
||||
from app.llm.client import LLMClient
|
||||
from app.agent.engine import Agent as AgentEngine
|
||||
from app.agent.tools import global_registry
|
||||
from app.agent.memory import ConversationMemory
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class WorkflowCreate(BaseModel):
|
||||
name: str
|
||||
description: str = ""
|
||||
steps: List[dict] = Field(default_factory=list)
|
||||
|
||||
|
||||
class WorkflowUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
steps: Optional[List[dict]] = None
|
||||
status: Optional[str] = None
|
||||
|
||||
|
||||
class WorkflowRunRequest(BaseModel):
|
||||
context: dict = Field(default_factory=dict)
|
||||
|
||||
|
||||
def _build_agent_engine_for_workflow(agent_id: str) -> AgentEngine:
|
||||
"""从数据库加载 Agent 并构建引擎"""
|
||||
session = get_session()
|
||||
try:
|
||||
from app.models import Agent as AgentModel
|
||||
agent = session.query(AgentModel).filter(AgentModel.id == agent_id).first()
|
||||
if not agent:
|
||||
raise ValueError(f"Agent {agent_id} not found")
|
||||
|
||||
config = get_config()
|
||||
provider_key = agent.llm_provider or "default"
|
||||
provider_cfg = config.llm_providers.get(provider_key) or config.default_llm
|
||||
|
||||
llm = LLMClient(provider_cfg)
|
||||
return AgentEngine(
|
||||
name=agent.name,
|
||||
llm_client=llm,
|
||||
tools=global_registry,
|
||||
memory=ConversationMemory(window_size=20),
|
||||
system_prompt=agent.system_prompt or "You are a helpful assistant.",
|
||||
temperature=agent.temperature,
|
||||
max_tokens=agent.max_tokens,
|
||||
)
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def list_workflows() -> List[dict]:
|
||||
session = get_session()
|
||||
try:
|
||||
workflows = session.query(Workflow).order_by(Workflow.created_at.desc()).all()
|
||||
return [
|
||||
{
|
||||
"id": w.id,
|
||||
"name": w.name,
|
||||
"description": w.description or "",
|
||||
"status": w.status,
|
||||
"step_count": len(w.definition.get("steps", [])) if w.definition else 0,
|
||||
"created_at": w.created_at,
|
||||
"updated_at": w.updated_at,
|
||||
}
|
||||
for w in workflows
|
||||
]
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.post("")
|
||||
async def create_workflow(data: WorkflowCreate) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
wf = Workflow(
|
||||
id=str(uuid.uuid4()),
|
||||
name=data.name,
|
||||
description=data.description,
|
||||
definition={"steps": data.steps},
|
||||
status="draft",
|
||||
)
|
||||
session.add(wf)
|
||||
session.commit()
|
||||
session.refresh(wf)
|
||||
return {
|
||||
"id": wf.id,
|
||||
"name": wf.name,
|
||||
"description": wf.description or "",
|
||||
"status": wf.status,
|
||||
"created_at": wf.created_at,
|
||||
}
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.get("/{workflow_id}")
|
||||
async def get_workflow(workflow_id: str) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
|
||||
if not wf:
|
||||
raise HTTPException(404, "Workflow not found")
|
||||
return {
|
||||
"id": wf.id,
|
||||
"name": wf.name,
|
||||
"description": wf.description or "",
|
||||
"definition": wf.definition or {},
|
||||
"status": wf.status,
|
||||
"created_at": wf.created_at,
|
||||
"updated_at": wf.updated_at,
|
||||
}
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.put("/{workflow_id}")
|
||||
async def update_workflow(workflow_id: str, data: WorkflowUpdate) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
|
||||
if not wf:
|
||||
raise HTTPException(404, "Workflow not found")
|
||||
|
||||
update_data = data.model_dump(exclude_unset=True)
|
||||
if "steps" in update_data and update_data["steps"] is not None:
|
||||
wf.definition = {"steps": update_data["steps"]}
|
||||
update_data.pop("steps")
|
||||
|
||||
for key, value in update_data.items():
|
||||
if value is not None:
|
||||
setattr(wf, key, value)
|
||||
|
||||
session.commit()
|
||||
session.refresh(wf)
|
||||
return {
|
||||
"id": wf.id,
|
||||
"name": wf.name,
|
||||
"description": wf.description or "",
|
||||
"status": wf.status,
|
||||
"updated_at": wf.updated_at,
|
||||
}
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.delete("/{workflow_id}")
|
||||
async def delete_workflow(workflow_id: str) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
|
||||
if not wf:
|
||||
raise HTTPException(404, "Workflow not found")
|
||||
session.delete(wf)
|
||||
session.commit()
|
||||
return {"deleted": workflow_id}
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.post("/{workflow_id}/run")
|
||||
async def run_workflow(workflow_id: str, data: WorkflowRunRequest, background_tasks: BackgroundTasks) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
|
||||
if not wf:
|
||||
raise HTTPException(404, "Workflow not found")
|
||||
|
||||
run = WorkflowRun(
|
||||
id=str(uuid.uuid4()),
|
||||
workflow_id=workflow_id,
|
||||
status="running",
|
||||
)
|
||||
session.add(run)
|
||||
session.commit()
|
||||
session.refresh(run)
|
||||
|
||||
# Build workflow model
|
||||
wf_model = Workflow(
|
||||
name=wf.name,
|
||||
description=wf.description or "",
|
||||
steps=[],
|
||||
)
|
||||
for step_data in (wf.definition or {}).get("steps", []):
|
||||
wf_model.steps.append(StepDef(**step_data))
|
||||
|
||||
# Start execution in background
|
||||
background_tasks.add_task(
|
||||
_execute_workflow,
|
||||
run.id,
|
||||
wf_model,
|
||||
data.context,
|
||||
)
|
||||
|
||||
return {"run_id": run.id, "status": "running"}
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
async def _execute_workflow(run_id: str, workflow: WorkflowModel, context: dict) -> None:
|
||||
"""后台执行工作流"""
|
||||
session = get_session()
|
||||
try:
|
||||
run = session.query(WorkflowRun).filter(WorkflowRun.id == run_id).first()
|
||||
if not run:
|
||||
return
|
||||
run.status = "running"
|
||||
session.commit()
|
||||
|
||||
# Build agent engines
|
||||
agent_engines = {}
|
||||
for step in workflow.steps:
|
||||
if step.agent_id and step.agent_id not in agent_engines:
|
||||
try:
|
||||
agent_engines[step.agent_id] = _build_agent_engine_for_workflow(step.agent_id)
|
||||
except Exception as e:
|
||||
run.status = "failed"
|
||||
run.error = f"Failed to load agent {step.agent_id}: {str(e)}"
|
||||
run.completed_at = datetime.utcnow()
|
||||
session.commit()
|
||||
return
|
||||
|
||||
# Simple step executor
|
||||
def step_executor(step_name: str, agent_id: str, prompt: str, step_context: dict) -> str:
|
||||
agent = agent_engines.get(agent_id)
|
||||
if not agent:
|
||||
return f"Error: Agent {agent_id} not available"
|
||||
messages = [{"role": "user", "content": prompt}]
|
||||
return agent.run(messages)
|
||||
|
||||
engine = WorkflowEngine()
|
||||
result = engine.run(workflow, context, step_executor=step_executor)
|
||||
|
||||
# Save step results
|
||||
for step_result in result.step_results:
|
||||
step_record = WorkflowStep(
|
||||
run_id=run_id,
|
||||
step_name=step_result.step_name,
|
||||
status=step_result.status.value,
|
||||
input_data={"context_keys": list(context.keys())},
|
||||
output_data={"output": step_result.output, "error": step_result.error},
|
||||
completed_at=datetime.utcnow(),
|
||||
)
|
||||
session.add(step_record)
|
||||
|
||||
run.status = "completed" if result.all_completed else "failed"
|
||||
run.results = result.to_dict()
|
||||
if result.any_failed:
|
||||
run.error = "One or more steps failed"
|
||||
run.completed_at = datetime.utcnow()
|
||||
session.commit()
|
||||
|
||||
except Exception as e:
|
||||
session.query(WorkflowRun).filter(WorkflowRun.id == run_id).update({
|
||||
"status": "failed",
|
||||
"error": str(e),
|
||||
"completed_at": datetime.utcnow(),
|
||||
})
|
||||
session.commit()
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.get("/{workflow_id}/runs")
|
||||
async def list_workflow_runs(workflow_id: str) -> List[dict]:
|
||||
session = get_session()
|
||||
try:
|
||||
runs = (
|
||||
session.query(WorkflowRun)
|
||||
.filter(WorkflowRun.workflow_id == workflow_id)
|
||||
.order_by(WorkflowRun.started_at.desc())
|
||||
.all()
|
||||
)
|
||||
return [
|
||||
{
|
||||
"id": r.id,
|
||||
"workflow_id": r.workflow_id,
|
||||
"status": r.status,
|
||||
"started_at": r.started_at,
|
||||
"completed_at": r.completed_at,
|
||||
"error": r.error,
|
||||
}
|
||||
for r in runs
|
||||
]
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@router.get("/runs/{run_id}")
|
||||
async def get_workflow_run(run_id: str) -> dict:
|
||||
session = get_session()
|
||||
try:
|
||||
run = session.query(WorkflowRun).filter(WorkflowRun.id == run_id).first()
|
||||
if not run:
|
||||
raise HTTPException(404, "Workflow run not found")
|
||||
|
||||
steps = [
|
||||
{
|
||||
"id": s.id,
|
||||
"step_name": s.step_name,
|
||||
"status": s.status,
|
||||
"input_data": s.input_data,
|
||||
"output_data": s.output_data,
|
||||
"started_at": s.started_at,
|
||||
"completed_at": s.completed_at,
|
||||
}
|
||||
for s in run.steps
|
||||
]
|
||||
|
||||
return {
|
||||
"id": run.id,
|
||||
"workflow_id": run.workflow_id,
|
||||
"status": run.status,
|
||||
"results": run.results,
|
||||
"error": run.error,
|
||||
"started_at": run.started_at,
|
||||
"completed_at": run.completed_at,
|
||||
"steps": steps,
|
||||
}
|
||||
finally:
|
||||
session.close()
|
||||
Reference in New Issue
Block a user