jhogyu/Liti-agent-Development

Fork 0

Files

T

jhogyu 0f22300489 First Commit - Creating Liti-agent Development

2026-07-02 17:08:27 +09:00

32 KiB

Raw Permalink Blame History

Agent Script YAML 작성 스킬 가이드

민사소송 원고 대리 에이전트 스크립트 작성 경험에서 축적된 패턴, 규칙, 주의사항 실행 백엔드: AgentBackend (FastAPI)

0. 백엔드 아키텍처 & 파일 저장 경로

0.1 user_id / workspace_id 기반 파일 격리

AgentBackend는 user_id와 workspace_id를 통해 사용자/워크스페이스별 파일 격리를 수행한다.

저장 경로 패턴:

{MCP_LOCALDOCS_PATH}/users/{SHA256(user_id)}/{SHA256(workspace_id)}/

조건	경로
user_id + workspace_id 둘 다 있음	`/mcp-localdocs/users/{hash_uid}/{hash_wid}/`
user_id만 있음	`/mcp-localdocs/users/{hash_uid}/`
둘 다 없음 (레거시)	`/mcp-localdocs/` (루트, 하위호환)

경로 결정: server.py::_resolve_base_dir(user_id, workspace_id)
user_id와 workspace_id는 SHA-256으로 해싱하여 디렉토리명으로 사용
파일 암호화: AES-256-GCM, LOCALDOCS_MASTER_SECRET에서 per-user 키 유도

0.2 user_id / workspace_id 전파 경로

클라이언트 요청 (user_id, workspace_id)
  ↓
server.py (REST/WebSocket/SSE 엔드포인트)
  ↓
Agent.run_with_confirmation(user_id, workspace_id)
  ↓
Stage.run() → stage.user_id, stage.workspace_id 설정
  ↓
MCPClientManager.from_dict(tools_config, user_id=..., workspace_id=...)
  ↓
MCP 서버 (localdocs) → _resolve_user(ctx) → 해시된 경로에서 파일 R/W

핵심: 에이전트 스크립트(YAML)에서는 user_id/workspace_id를 직접 다루지 않는다. 백엔드가 자동으로 전파한다. code-executor에서 직접 localdocs MCP를 호출할 때도 세션 초기화 시 자동으로 user/workspace 컨텍스트가 바인딩된다.

0.3 Default_Agent 폴더

Default_Agent/ 폴더는 워크스페이스 초기화 시 pre_loads/에서 자동 복사되는 공용 규칙 파일 저장소다.

파일	용도
`Keywords_Criteria.txt`	판례 검색 키워드 기준
`Mapping_Rule_Case_DB.txt`	판례 DB 매핑 규칙
`청구취지작성규칙_Mapping_Table.md`	case_kind → 작성규칙 매핑
`complaint_template.hwpx`	소장 hwpx 템플릿
`Global_Rules.md`	글로벌 규칙

0.4 환경 변수

변수	기본값	설명
`MCP_LOCALDOCS_PATH`	`/mcp-localdocs`	localdocs 루트 경로
`LOCALDOCS_MASTER_SECRET`	(필수)	파일 암호화 마스터 시크릿
`USER_STORAGE_LIMIT_MB`	`500`	유저당 저장 한도

1. 전체 구조

Agent:
  name: <에이전트명>
  description: <설명>
  version: <버전>

  Stages:
    - name: <스테이지명>
      description: <설명>
      tools:
        mcpServers:
          <서버명>:
            type: streamable-http
            url: "<URL>"
            description: <설명>
            headers:              # 선택
              Authorization: Bearer <token>

      # 둘 중 하나 선택: map_reduce 또는 task_procedure
      map_reduce: { ... }
      # 또는
      task_procedure: { ... }
      tasks: [ ... ]

Stage 옵션

필드	타입	기본값	설명
`skip_confirm`	bool	`false`	`true`로 설정하면 실행 확인 없이 바로 실행 (스킵 방지)

- name: stage_build
  description: 전처리 파일 생성
  skip_confirm: true          # 확인 없이 바로 실행
  prevs: []
  nexts: [stage_main]

Stage 간 의존성 (DAG)

Stages:
  - name: stage_A
    nexts: [stage_B, stage_C]
  - name: stage_B
    prevs: [stage_A]
    nexts: [stage_D]
  - name: stage_D
    prevs: [stage_B, stage_C]   # 모두 완료되어야 실행

2. 실행 모드

2.1 MapReduce — 동일 작업을 다수 아이템에 병렬 실행

map_reduce:
  max_concurrency: 12
  source:
    mcp:
      server: localdocs
      tool_name: read_docs
      parameters:
        doc_names: ["claims_identified.json"]
      key: ["identified_claims"]       # JSON 배열 경로

  shared_context:                       # 선택: 1회 로드, 모든 task에서 {{shared.X}}
    - name: keywords_criteria
      mcp:
        server: localdocs
        tool_name: read_docs
        parameters: { doc_names: ["Keywords_Criteria.txt"] }

  map:
    tasks:
      - task_name: step1 ...
      - task_name: step2 ...           # 아이템당 순차 실행

  reduce:
    - task_name: aggregate ...          # map 완료 후 1회 실행

핵심:

max_concurrency개 아이템이 동시에 map 파이프라인 실행
각 아이템 내 task는 순차 실행
reduce는 모든 map 완료 후 1회 실행
Stage 레벨 llm_provider/llm_model은 map_reduce에서 불필요 (각 task가 자체 지정)

2.2 Task Procedure — DAG 기반 분기/합류

task_procedure:
  IN:
    nexts: [Task_A]
  Task_A:
    nexts:
      - task: Task_B
        when: "output.score >= 80"
      - task: Task_C
        when: "output.score < 80"
  Task_B:
    nexts: [OUT]
    wait_until: [Task_A]
  Task_C:
    nexts: [OUT]
    wait_until: [Task_A]
  OUT:
    nexts: []
    wait_until: [Task_B, Task_C]

2.3 Wildcard Fan-out — 런타임 동적 인스턴스 생성

부모 task 의 stdout JSON 에 {"dynamic_fanout": [item0, item1, ...]} 가 들어 있으면, task_procedure 의 Task_X_* (별 * 접미사) 패턴이 Task_X_0, Task_X_1, ... 로 런타임 확장된다. 각 인스턴스는 자기 item 을 {{item.*}} 로 받는다. (agent.py:1304-1351, _is_wildcard_name / _expand_wildcard)

Fan-out 토큰 3종

토큰	의미	사용처
`Task_X_*`	wildcard 정의. 인스턴스로 확장될 task template	`tasks[].task_name`, `task_procedure` key
`Task_Y_{same_ordinal}`	자기 인덱스 (ordinal) 로 치환되는 cross-fan-out 참조	`nexts` / `wait_until` 항목
`all Task_X_*`	모든 인스턴스 완료를 한 번에 대기하는 집계 토큰	`wait_until` 항목

전형적 fan-out DAG

task_procedure:
  IN:
    nexts: [Task_Planner]
    wait_until: []
  Task_Planner:
    nexts: [Task_B1_map_doc_*]
    wait_until: [IN]
  Task_B1_map_doc_*:                          # 부모의 dynamic_fanout 길이만큼 확장
    nexts:
      - "Task_B2_map_events_{same_ordinal}"   # 같은 ordinal 의 B2 가 wait
      - "Task_B1_quality_gate"                # 정적 집계자
    wait_until: [Task_Planner]
  Task_B2_map_events_*:
    nexts: [Task_B2_quality_gate]
    wait_until:
      - "Task_B1_map_doc_{same_ordinal}"      # cross-fan-out 동일 ordinal 대기
  Task_B1_quality_gate:
    nexts: [Task_B2_quality_gate]
    wait_until: ["all Task_B1_map_doc_*"]     # 전체 인스턴스 완료 대기
  Task_B2_quality_gate:
    nexts: [OUT]
    wait_until:
      - "all Task_B2_map_events_*"
      - Task_B1_quality_gate

부모(planner) task 의 stdout 규약

Task_Planner (code-executor 또는 LLM task) 가 다음 JSON 한 줄을 stdout 으로 출력해야 wildcard 확장이 발동된다:

print(json.dumps({
    "dynamic_fanout": [
        {"assigned_ordinal": 1, "ordinal_label": "001", "evidence_index_candidate": "E-001",
         "shard_doc_path": "evidence_shards/E-001.json", ...},
        {"assigned_ordinal": 2, "ordinal_label": "002", ...},
        ...
    ],
    "planner_status": "READY",
    "mapper_execution_allowed": True
}, ensure_ascii=False))

각 인스턴스의 prompt / code 에서 {{item.assigned_ordinal}}, {{item.shard_doc_path}} 등으로 접근.

Wildcard task 의 task_config

- task_name: Task_B1_map_doc_*
  max_concurrency: 12                         # 인스턴스 병렬 상한 (없으면 무제한)
  preflight_files:                            # 인스턴스마다 다른 path 가능
    - evidence_shard_plan.json                # 정적 (모든 인스턴스 공통, cache 됨)
    - "{{item.shard_doc_path}}"               # 동적 (인스턴스별)
    - client_meeting.md
  llm_provider: google
  llm_model: gemini-3.1-flash-lite-preview
  prompts:
    - role: user
      content: |-
        <ASSIGNED_ORDINAL>{{item.assigned_ordinal}}</ASSIGNED_ORDINAL>
        본 task 는 `{{item.shard_doc_path}}` 1건만 처리한다.
        출력 경로: evidence_indexed_parts/E-{{item.ordinal_label}}.json

주의:

max_concurrency: N → semaphore 로 N 개 instance 만 동시 실행. 외부 API rate limit (예: Gemini 분당 토큰 한도) 회피용
wildcard task 의 task_name 은 반드시 _* 로 끝나야 함 (_is_wildcard_name)
cross-fan-out 참조 시 {same_ordinal} 은 같은 ordinal 끼리만 연결. 다른 ordinal 의 동일 task type 은 독립.
aggregate token all Task_X_* 은 모든 instance 완료를 한 번에 대기. quality gate 패턴의 표준 형태.
자세한 동작 흐름: agent.py::TaskProcedureExecutor._maybe_expand_wildcards

3. Task 유형

3.1 LLM Task

- task_name: analyze
  llm_provider: google              # google | anthropic | openai | "https://<endpoint>" (OpenAI-compatible)
  llm_model: gemini-3.1-pro-preview
  llm_reasoning: high               # low | medium | high
  llm_verbosity: medium             # low | medium | high
  llm_endpoint: completions         # "completions" (default) | "responses"
  use_tools: ["localdocs"]          # LLM이 사용할 MCP 서버
  cache_control:                    # 선택 (Gemini/Anthropic)
    mode: auto                      # "auto" | "explicit"
    ttl: 1h                         # 5m / 10m / 1h / 2h ... (Pro 모델 long-running 은 1h+ 권장)
  preflight: true                   # default true. 프롬프트의 파일명을 미리 read_docs (LLM-based task 만 적용)
  preflight_files:                  # 선택. 명시 시 regex 스캔 대신 이 화이트리스트만 preflight
    - evidence_shard_plan.json      # glob (*, ?) 지원
    - "evidence_indexed_parts/E-*.json"
    - client_meeting.md
  llm_history_continuous: prev_task_name  # 선택. 이전 task/stage 의 response_id 로 체이닝 (Responses API)
  prompts:
    - role: system
      content: |
        <system_role>...</system_role>
    - role: user
      content: |
        {{prev.load_data.result}}

주의:

use_tools 를 주면 LLM 이 도구 호출 가능 → 프롬프트에서 허용 범위를 명시해야 함
llm_reasoning 은 Google native SDK 경로에서만 작동 (cache_control 없이도 가능)
preflight: false 명시하면 prompt 본문 regex 스캔과 자동 read_docs 모두 비활성화 (forbidden-files 정책이 강한 task 에서 권장)
preflight_files 의 entry 에 {{item.*}} 같은 동적 path 가 있으면 fan-out 인스턴스별로 다르게 로드. 정적 path 만으로 구성된 entry 는 instance-invariant 로 판단되어 Gemini cache 의 user_prefix 영역에 inline 됨 (다른 인스턴스/task 와 cache 공유)

3.2 Direct MCP Task (LLM 없이 도구 직접 호출)

- task_name: search_weaviate
  mcp: weaviate
  tool_name: search_hybrid
  parameters:
    collection_name: "{{prev.generate_query.search_params.collection_name}}"
    tenant: "{{prev.generate_query.search_params.tenant}}"
    query: "{{prev.generate_query.search_params.query}}"
    alpha: "{{prev.generate_query.search_params.alpha}}"
    limit: "{{prev.generate_query.search_params.limit}}"

3.3 Code-Executor Task (인라인 Python)

- task_name: save_results
  mcp: code-executor
  tool_name: run_code
  parameters:
    language: python
    requirements: "httpx"
    network: "agent-network"
    timeout: 60
    code: |
      #!/usr/bin/env python3
      import json, sys
      # ... 코드
      print(json.dumps({"status": "ok"}))   # stdout = task output

3.4 조건부 Task (map 전용)

- task_name: search_if_ready
  when: "prev.generate_query.query_generated == true"
  mcp: weaviate
  tool_name: search_hybrid
  parameters: { ... }

4. 템플릿 변수

변수	사용 위치	설명
`{{item.X}}`	map task / wildcard fan-out instance	현재 반복/인스턴스 아이템의 필드 (e.g., `{{item.assigned_ordinal}}`)
`{{item}}`	map / fan-out	항목 전체 (dict/list 면 자동 JSON 직렬화)
`{{item_json}}`	map / fan-out	항목 전체 JSON 문자열
`{{item_index}}`	map / fan-out	0-based 인덱스
`{{ordinal}}`	wildcard fan-out instance	인스턴스 ordinal (0-based, `item_index` 와 동일)
`{{prev.task_name}}`	모든 task	이전 task 의 전체 출력 (stdout JSON 자동 파싱)
`{{prev.task_name.field}}`	모든 task	이전 task 출력의 특정 필드
`{{prev.task_name.nested.field}}`	모든 task	중첩 필드 접근 (dot-notation)
`{{stages.<stage_name>}}`	stage 간	이전 stage 의 output 전체 (`apply_stage_context`)
`{{stages.<stage_name>.field}}`	stage 간	이전 stage output 의 특정 필드
`{{map_results}}`	reduce task	map 전체 결과 (JSON 문자열)
`{{map_results_b64}}`	reduce task	map 전체 결과 (base64 인코딩)
`{{map_results_count}}`	reduce task	map 결과 건수
`{{shared.name}}`	map/reduce	shared_context 로 로드한 데이터
`{{__user_hash__}}`	code-executor task	현재 user_id 의 SHA-256 (clientInfo 주입용, 자동 substitute)
`{{__workspace_hash__}}`	code-executor task	현재 workspace_id 의 SHA-256

fan-out 전용 토큰 (task_procedure 안의 nexts / wait_until 항목에서만 의미):

토큰	위치	설명
`Task_X_*`	task_procedure key, nexts/wait_until	wildcard task template — 부모의 `dynamic_fanout` 길이만큼 인스턴스 확장
`Task_X_{same_ordinal}`	nexts/wait_until 항목	자기 인스턴스 ordinal 로 치환되어 cross-fan-out 동일 ordinal 끼리 연결
`all Task_X_*`	wait_until 항목	wildcard 의 모든 인스턴스 완료를 한 번에 대기 (집계자 / quality gate 패턴)

5. MCP Localdocs 보일러플레이트

code-executor에서 localdocs를 사용하는 모든 Python 코드에 필요한 공통 패턴:

#!/usr/bin/env python3
import itertools, json, sys
import httpx

LOCALDOCS_URL = "http://mcp-localdocs:8012/mcp"
MCP_HEADERS = {"Content-Type": "application/json",
               "Accept": "application/json, text/event-stream"}
CLIENT = httpx.Client(timeout=60)
MSG_ID_COUNTER = itertools.count(10)

def next_msg_id(): return next(MSG_ID_COUNTER)

def _init():
    r = CLIENT.post(LOCALDOCS_URL, json={
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {"protocolVersion": "2025-03-26", "capabilities": {},
                   "clientInfo": {"name": "<task_name>", "version": "1.0",
                                "user_id": "{{__user_hash__}}", "workspace_id": "{{__workspace_hash__}}"}}
    }, headers=MCP_HEADERS)
    r.raise_for_status()
    sid = r.headers.get("mcp-session-id")
    if sid: MCP_HEADERS["mcp-session-id"] = sid
    CLIENT.post(LOCALDOCS_URL,
                json={"jsonrpc": "2.0", "method": "notifications/initialized"},
                headers=MCP_HEADERS).raise_for_status()

def _parse_mcp(text):
    for line in text.strip().split("\n"):
        if line.startswith("data: "):
            try: return json.loads(line[6:])
            except: return None
    try: return json.loads(text)
    except: return None

def _call(name, args, mid):
    r = CLIENT.post(LOCALDOCS_URL, json={
        "jsonrpc": "2.0", "id": mid, "method": "tools/call",
        "params": {"name": name, "arguments": args}
    }, headers=MCP_HEADERS)
    r.raise_for_status()
    p = _parse_mcp(r.text)
    if not p or "result" not in p:
        raise RuntimeError(f"MCP {name} failed")
    return p

def unwrap_text(p, name):
    text = (p["result"].get("content") or [{}])[0].get("text", "")
    if not text: raise RuntimeError(f"Empty response: {name}")
    try:
        outer = json.loads(text)
    except:
        return text
    if isinstance(outer, dict) and "results" in outer:
        r0 = (outer.get("results") or [{}])[0]
        inner = r0.get("content") or r0.get("text") or ""
        if not inner: raise RuntimeError(f"Empty content: {name}")
        return inner if isinstance(inner, str) else json.dumps(inner, ensure_ascii=False)
    return json.dumps(outer, ensure_ascii=False) if not isinstance(outer, str) else outer

def read_text(name):
    return unwrap_text(_call("read_docs", {"doc_names": [name]}, next_msg_id()), name)

def read_json(name):
    raw = read_text(name)
    return json.loads(raw) if isinstance(raw, str) else raw

def write_doc(path, content):
    _call("write_file", {"path": path, "content": content, "overwrite": True}, next_msg_id())

_init()
# ... 작업 수행 ...
print(json.dumps({"status": "ok", ...}))

5.1 바이너리 파일 읽기/쓰기

localdocs는 텍스트와 바이너리를 별도 MCP tool로 처리한다:

도구	파라미터	용도
`read_docs`	`{"doc_names": ["file.json"]}`	텍스트 파일 읽기
`read_binary_doc`	`{"doc_name": "file.hwpx"}`	바이너리 파일 읽기
`write_file`	`{"path": "file.json", "content": "...", "overwrite": true}`	텍스트 파일 저장
`write_binary_file`	`{"path": "file.hwpx", "content_base64": "...", "overwrite": true}`	바이너리 파일 저장
`list_docs`	`{"path": "folder", "pattern": "*"}`	파일 목록 조회
`list_folders`	`{}`	폴더 목록 조회

바이너리 읽기 응답 구조:

# read_binary_doc 응답의 text 필드가 JSON envelope
envelope = json.loads(data_str)
binary_bytes = base64.b64decode(envelope["content_base64"])

바이너리 쓰기:

encoded = base64.b64encode(data).decode("ascii")
_call("write_binary_file", {
    "path": "소장.hwpx",
    "content_base64": encoded,
    "overwrite": True,
}, next_msg_id())

5.2 Direct MCP에서 user_id / workspace_id 컨텍스트 전달

문제: code-executor에서 localdocs를 직접 호출하면 파일이 안 보인다

AgentBackend가 Stage를 실행할 때 MCPClientManager를 통해 localdocs에 연결하면, clientInfo에 SHA-256 해시된 user_id와 workspace_id가 자동으로 포함된다. localdocs 서버는 이 값으로 저장 경로를 결정한다.

그러나 code-executor task 안의 Python 코드가 localdocs에 직접 HTTP 요청을 보내면, clientInfo에 user/workspace 정보가 없으므로 **루트 경로(DOCS_DIR)**에 접근하게 된다. 이 경우 사용자의 파일이 보이지 않는다.

MCPClientManager가 전달하는 방식 (자동)

# llm_bridge/client.py — MCPClientManager가 세션 초기화 시 자동 수행
hashed_uid = hashlib.sha256(user_id.encode("utf-8")).hexdigest()
hashed_wid = hashlib.sha256(workspace_id.encode("utf-8")).hexdigest()
client_info = Implementation(
    name="eroom-agent", version="1.0",
    user_id=hashed_uid,
    workspace_id=hashed_wid,
)
session = ClientSession(*stream_context, client_info=client_info)

localdocs 서버가 읽는 방식

# mcp_localdocs_server.py::_resolve_user(ctx)
hashed_uid = getattr(client_info, 'user_id', None)   # SHA-256 hex
cid = getattr(client_info, 'workspace_id', None)      # SHA-256 hex
# → {DOCS_DIR}/users/{hashed_uid}/{cid}/

code-executor에서 직접 호출할 때의 경로

clientInfo에 user_id 포함	접근 경로
❌ 미포함 (현재 보일러플레이트)	`/mcp-localdocs/` (루트 — 다른 사용자 파일과 혼재)
✅ 포함	`/mcp-localdocs/users/{hash_uid}/{hash_wid}/` (격리됨)

해결: code-executor 보일러플레이트에 user/workspace 전달

현재 code-executor 내 Python 코드는 JSON-RPC initialize 메서드의 clientInfo에 user_id/workspace_id를 넣지 않는다. 이를 전달하려면:

# ✅ user_id/workspace_id를 clientInfo에 포함시키는 방식
def _init():
    r = CLIENT.post(LOCALDOCS_URL, json={
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {
                "name": "<task_name>",
                "version": "1.0",
                "user_id": "<SHA-256 hashed user_id>",       # 필수
                "workspace_id": "<SHA-256 hashed workspace_id>"  # 선택
            }
        }
    }, headers=MCP_HEADERS)
    ...

그러나 code-executor의 Python 코드에서는 원본 user_id를 알 수 없다. 이 값은 AgentBackend가 관리하며, YAML 스크립트에 노출되지 않는다.

해결: `{{__user_hash}}` / `{{workspace_hash__}}` 템플릿 변수

AgentBackend가 {{__user_hash__}}와 {{__workspace_hash__}}를 자동으로 SHA-256 해시값으로 렌더링한다. code-executor 코드의 clientInfo에 넣으면 localdocs가 user 경로에서 파일을 읽고 쓴다.

# ✅ clientInfo에 user_id/workspace_id를 반드시 포함
def _init():
    r = CLIENT.post(LOCALDOCS_URL, json={
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {
                "name": "<task_name>",
                "version": "1.0",
                "user_id": "{{__user_hash__}}",
                "workspace_id": "{{__workspace_hash__}}"
            }
        }
    }, headers=MCP_HEADERS)

clientInfo에 user_id/workspace_id가 없으면 localdocs는 루트 경로에서 파일을 찾으므로, 멀티유저 환경에서 파일을 못 읽거나 다른 사용자 파일에 접근하는 문제가 발생한다.

localdocs read_docs 응답의 Extra data 문제

localdocs read_docs 응답에서 content 필드 뒤에 "path": "파일명" 등 추가 데이터가 붙어 json.loads가 Extra data 에러로 실패하는 경우가 있다.

# ❌ BAD — Extra data로 실패 가능
def read_json(name):
    raw = _unwrap(_call("read_docs", {"doc_names": [name]}), name)
    return json.loads(raw)

# ✅ GOOD — raw_decode fallback으로 첫 번째 JSON 객체만 파싱
def read_json(name):
    raw = _unwrap(_call("read_docs", {"doc_names": [name]}), name)
    if not isinstance(raw, str): return raw
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        obj, _ = json.JSONDecoder().raw_decode(raw.strip())
        return obj

이 패턴은 _extract_json_from_read_docs_result 같은 래퍼에도 동일하게 적용해야 한다:

# inner가 JSON 문자열일 때
try:
    return json.loads(inner)
except json.JSONDecodeError:
    try:
        obj, _ = json.JSONDecoder().raw_decode(inner.strip())
        return obj
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Content of {doc_name} is not valid JSON: {inner[:300]}") from exc

6. 주의사항 & 함정 (Pitfalls)

6.1 Python 문자열 리터럴에 템플릿 삽입

# ❌ BAD — Python이 \n, \" 등을 escape 처리하여 JSON 파괴
_RAW = """{{prev.search_weaviate}}"""

# ✅ GOOD — raw string으로 escape 처리 방지
_RAW = r"""{{prev.search_weaviate}}"""

이유: 템플릿 확장 후 Python이 """...""" 를 파싱할 때 \n → 줄바꿈, \" → " 변환 발생. raw string r"""..."""은 이를 방지.

6.2 Reduce에서 map_results 처리

# ❌ BAD — 큰 결과에서 raw string 삽입 시 깨질 수 있음
_RAW = r"""{{map_results}}"""

# ✅ GOOD — base64 디코딩
import base64
_RAW = base64.b64decode("{{map_results_b64}}").decode("utf-8")
results = json.loads(_RAW)

6.3 map_results 구조

reduce에서 받는 map_results의 각 아이템 구조:

{
  "item_index": 0,
  "item": { "claim_id": "C-001", ... },
  "task_results": {
    "task_name_1": "output...",
    "task_name_2": "output..."
  }
}

claim_id 추출 패턴:

def extract_claim_id(item):
    if "item" in item and isinstance(item["item"], dict):
        return item["item"].get("claim_id")
    return item.get("claim_id")

6.4 LLM이 도구를 오용하는 경우

use_tools를 주면 LLM이 write_file까지 호출할 수 있음:

# ❌ LLM이 파일 저장까지 수행하고 "Successfully saved..." 반환
use_tools: ["localdocs"]
# 프롬프트에 제약 없음

# ✅ 프롬프트에 명시적 제약 추가
# "도구 호출은 오직 read_docs만 허용. write_file 금지."

6.5 Gemini 제약

조합	결과
`cache_control` only	✅ native SDK
`reasoning` only (no `cache_control`)	✅ native SDK, ThinkingConfig 적용
`reasoning` + `cache_control`	✅ 작동 (캐시 + thinking)
`cache_control` + `tools`	✅ 작동 (llm_bridge 가 tools 도 cache 에 저장, request 에서 omit)
`reasoning` + `cache_control` + `tools`	✅ 작동 (위와 동일 경로)

참고: Gemini API 자체는 cached_content 사용 시 같은 request 에서 system_instruction/tools/tool_config 를 다시 보낼 수 없다. llm_bridge 의 _ensure_gemini_cache 가 tools 도 cache 안에 함께 저장하고 generate request 에서는 omit 하므로 yaml 작성자는 cache + tools 를 자유롭게 같이 쓸 수 있다 (bridge.py:309-316).

6.5.1 Cache TTL 만료 처리

Pro 모델 (gemini-3.1-pro-preview 등) 의 long-running 호출이 TTL 을 넘기면 다음 호출에서 403 PERMISSION_DENIED. CachedContent not found 가 발생한다. llm_bridge 는 이를 감지해 cache 를 자동 무효화하고 1회 재시도한다. yaml 작성 시:

Pro 모델 + reasoning=high 조합은 ttl: 1h 이상 권장 (5m / 10m 은 짧다)
Flash / Flash-lite + 짧은 prompt 는 ttl: 10m 정도로도 충분

6.6 파일 병합/결합 시 내용 생략 금지

여러 입력 파일을 하나의 combined YAML로 합칠 때, 긴 프롬프트나 코드 블록을 절대 placeholder로 대체하지 말라.

# ❌ BAD — 내용을 placeholder 주석으로 생략
prompts:
  - role: user
    content: |
      [Task_B prompt content preserved from file 1 - lines 1068-1631]
      This is a comprehensive LLM task for claims identification.

# ✅ GOOD — 원본 내용 전체를 그대로 삽입
prompts:
  - role: user
    content: |
      <COMMON_CACHE_PREFIX_V0>
      당신은 대한민국 민사소송 실무를 지원하는 ...
      (원본 전체 내용)
      </TASK_B_DYNAMIC_TAIL_V5>

6.7 LLM 출력 파싱

LLM이 JSON을 마크다운 코드블록으로 감쌀 수 있음:

def parse_llm_json(raw):
    s = raw.strip()
    if s.startswith("```"):
        parts = s.split("```")
        for p in parts:
            p = p.strip()
            if p.startswith("json"): p = p[4:].strip()
            if p.startswith("{") or p.startswith("["): s = p; break
    try: return json.loads(s)
    except json.JSONDecodeError: pass
    fixed = s.replace('\n', '\\n').replace('\r', '\\r').replace('\t', '\\t')
    try: return json.loads(fixed)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Parse failed: {e} | {repr(s[:150])}")

6.8 한글 파일명 유니코드 정규화

macOS는 NFD, Linux/서버는 NFC를 사용한다. localdocs에서 한글 파일명이 "not found"가 되는 주요 원인:

# ✅ 한글 파일명은 NFC 정규화 후 사용
import unicodedata
doc_name = unicodedata.normalize("NFC", "청구취지작성규칙_대여금청구.md")

# ✅ 또는 영문 파일명 사용 (가장 안전)
TEMPLATE_DOC = "Default_Agent/complaint_template.hwpx"

7. 파일 네이밍 컨벤션

패턴	예시	설명
`C-###`	C-001	청구권 ID
`F-###`	F-012	사실(Fact) ID
`E-###`	E-005	증거(Evidence) ID
`C-###_<설명>.json`	C-001_claim_description.json	청구권별 데이터
`C-###_<설명>.md`	C-001_defense_rebuttal.md	청구권별 분석 결과

8. 프롬프트 구조 패턴

prompts:
  - role: system
    content: |
      <system_role>역할 정의</system_role>

      <input_output_policy>
      IN: 입력 파일/데이터 명시
      OUT: 출력 형식 명시 (파일 저장 금지 등)
      </input_output_policy>

      <constraints>
      [제약1]: 설명
      [제약2]: 설명
      </constraints>

      <algorithm_to_perform>
      [Step 1]: ...
      [Step 2]: ...
      </algorithm_to_perform>

      <output_contract>
      JSON Schema 또는 Markdown 템플릿
      </output_contract>

  - role: user
    content: |
      ## 처리할 데이터:
      {{prev.load_data.content}}

9. 자주 사용하는 MCP 서버

서버	URL	주요 도구
localdocs	`http://mcp-localdocs:8012/mcp`	`read_docs`, `read_binary_doc`, `write_file`, `write_binary_file`, `list_docs`, `list_folders`
weaviate	`https://weaviate.eroomai.com/mcp`	`search_hybrid`
code-executor	`https://code-executor.mcp.eroomai.com/mcp`	`run_code`

10. HWPX 소장 생성 패턴

10.1 HWPX 파일 구조

hwpx는 ZIP 아카이브이며 내부에 XML 파일들이 있다:

파일	역할
`Contents/header.xml`	문단 속성 정의 (`hh:paraPr` — 글꼴, 여백, 줄간격 등)
`Contents/section0.xml`	실제 문단 내용 (`hp:p` — 텍스트, 테이블 등)

10.2 문단 들여쓰기 (paraPr)

section0.xml의 각 <hp:p> 태그는 paraPrIDRef로 header.xml의 <hh:paraPr> 를 참조한다:

<!-- section0.xml -->
<hp:p paraPrIDRef="47" ...>
  <hp:run charPrIDRef="0"><hp:t>(1) 원고 우방캐피탈...</hp:t></hp:run>
</hp:p>

<!-- header.xml -->
<hh:paraPr id="47" ...>
  <hh:margin>
    <hc:left value="4000" unit="HWPUNIT"/>   <!-- 40pt 들여쓰기 -->
  </hh:margin>
</hh:paraPr>

새 paraPr 추가 시 필수:

기존 최대 ID + 1로 순차 ID 할당
<hh:paraProperties itemCnt="N"> 의 itemCnt 값 업데이트 (N → N+1)
itemCnt가 실제 항목 수와 불일치하면 뷰어가 추가 항목을 무시함

10.3 HWPX 단위

HWPUNIT	환산
100	1pt
7200	1inch
~283.5	1mm

11. 체크리스트: 새 스크립트 작성 시

Stage 레벨 llm_provider/llm_model: map_reduce에서는 불필요
각 LLM task에 provider/model/reasoning 개별 지정
code-executor에서 localdocs 사용 시 full boilerplate 포함 + clientInfo 에 {{__user_hash__}}/{{__workspace_hash__}} 필수
템플릿 변수를 Python에 삽입할 때 r"""...""" 사용
reduce에서 map_results는 {{map_results_b64}} + base64 디코드
use_tools를 주는 LLM task → 프롬프트에 허용/금지 범위 명시
Weaviate 검색 결과 파싱: r"""...""" + JSON 제어문자 escape 처리
LLM 출력: 코드블록 래핑 대비 파싱 로직
map_results 아이템 구조: item.item.claim_id 형태 확인
Pro 모델 + reasoning=high → cache_control.ttl: 1h 이상 (만료 시 자동 재시도되지만 cache 없는 호출은 비용↑)
파일 병합 시 프롬프트/코드를 placeholder나 요약 주석으로 생략하지 않았는지 확인
한글 파일명: NFC 정규화 또는 영문명 사용
바이너리 파일: read_binary_doc / write_binary_file 사용 (텍스트 도구와 혼용 금지)
HWPX 수정 시: header.xml itemCnt 동기화 필수
확인 없이 실행해야 하는 Stage: skip_confirm: true 설정
Wildcard fan-out task: 이름이 _* 로 끝남 + 부모가 stdout 으로 {"dynamic_fanout": [...]} JSON 출력
Fan-out 인스턴스의 wait_until 에 cross-fan-out 동일 ordinal 참조 시 Task_X_{same_ordinal} 사용
Quality gate 처럼 fan-out 전체 완료를 대기하는 집계자는 wait_until: ["all Task_X_*"]
Fan-out 인스턴스의 외부 API rate limit 방지: task 에 max_concurrency: N
Fan-out 인스턴스의 preflight_files 항목 중 정적 path 는 자동으로 cache 공유 영역에 inline 됨 (인스턴스별 path 는 {{item.*}} 사용)
LLM-based task 에 외부 도구 호출 금지 정책이 있으면 preflight: false 명시

32 KiB Raw Permalink Blame History